diff --git a/Cargo.lock b/Cargo.lock index 69e1f06..3443c13 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -120,7 +120,7 @@ checksum = "9035ad2d096bed7955a320ee7e2230574d28fd3c3a0f186cbea1ff3c7eed5dbb" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -217,7 +217,7 @@ checksum = "7aa268c23bfbbd2c4363b9cd302a4f504fb2a9dfe7e3451d66f35dd392e20aca" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -361,7 +361,7 @@ dependencies = [ "heck", "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -518,7 +518,7 @@ checksum = "1ac70aa55017e108007fbaf5aa0f54b021c98f92ff8af59d42eda9da96e3dd4f" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -776,6 +776,31 @@ dependencies = [ "wasm-bindgen", ] +[[package]] +name = "hard-xml" +version = "1.41.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3b07b8ba970e18a03dbb79f6786b6e4d6f198a0ac839aa5182017001bb8dee17" +dependencies = [ + "hard-xml-derive", + "jetscii", + "lazy_static", + "memchr", + "xmlparser", +] + +[[package]] +name = "hard-xml-derive" +version = "1.41.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f0c43e7c3212bd992c11b6b9796563388170950521ae8487f5cdf6f6e792f1c8" +dependencies = [ + "bitflags", + "proc-macro2", + "quote", + "syn 1.0.109", +] + [[package]] name = "hashbrown" version = "0.15.5" @@ -1106,6 +1131,12 @@ version = "1.0.18" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8f42a60cbdf9a97f5d2305f08a87dc4e09308d1276d28c869c684d7777685682" +[[package]] +name = "jetscii" +version = "0.5.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "47f142fe24a9c9944451e8349de0a56af5f3e7226dc46f3ed4d4ecc0b85af75e" + [[package]] name = "js-sys" version = "0.3.103" @@ -1267,9 +1298,10 @@ dependencies = [ "newsfeed-core", "newsfeed-data", "newsfeed-entities", + "newsfeed-fetch", "serde", "serde_json", - "thiserror", + "thiserror 2.0.18", "tokio", "tower", "tower-http", @@ -1290,7 +1322,7 @@ dependencies = [ "rand 0.8.6", "serde", "sha2", - "thiserror", + "thiserror 2.0.18", "uuid", ] @@ -1317,11 +1349,28 @@ dependencies = [ "chrono", "serde", "serde_json", - "thiserror", + "thiserror 2.0.18", "ts-rs", "uuid", ] +[[package]] +name = "newsfeed-fetch" +version = "0.1.0" +dependencies = [ + "anyhow", + "async-trait", + "chrono", + "feed-rs", + "newsfeed-core", + "newsfeed-entities", + "opml", + "regex", + "reqwest", + "tokio", + "url", +] + [[package]] name = "newsfeed-worker" version = "0.1.0" @@ -1329,11 +1378,11 @@ dependencies = [ "anyhow", "chrono", "clap", - "feed-rs", "figment", "newsfeed-core", "newsfeed-data", "newsfeed-entities", + "newsfeed-fetch", "reqwest", "serde", "tokio", @@ -1408,6 +1457,17 @@ version = "1.70.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "384b8ab6d37215f3c5301a95a4accb5d64aa607f1fcb26a11b5303878451b4fe" +[[package]] +name = "opml" +version = "1.1.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "df2f96426c857a92676dc29a9e2a181eb39321047ac994491c69eae01619ddf2" +dependencies = [ + "hard-xml", + "serde", + "thiserror 1.0.69", +] + [[package]] name = "parking" version = "2.2.1" @@ -1468,7 +1528,7 @@ dependencies = [ "proc-macro2", "proc-macro2-diagnostics", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -1560,7 +1620,7 @@ checksum = "af066a9c399a26e020ada66a034357a868728e72cd426f3adcd35f80d88d88c8" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", "version_check", "yansi", ] @@ -1589,7 +1649,7 @@ dependencies = [ "rustc-hash", "rustls", "socket2", - "thiserror", + "thiserror 2.0.18", "tokio", "tracing", "web-time", @@ -1611,7 +1671,7 @@ dependencies = [ "rustls", "rustls-pki-types", "slab", - "thiserror", + "thiserror 2.0.18", "tinyvec", "tracing", "web-time", @@ -1942,7 +2002,7 @@ checksum = "d540f220d3187173da220f885ab66608367b6574e925011a9353e4badda91d79" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -2145,7 +2205,7 @@ dependencies = [ "serde_json", "sha2", "smallvec", - "thiserror", + "thiserror 2.0.18", "tokio", "tokio-stream", "tracing", @@ -2164,7 +2224,7 @@ dependencies = [ "quote", "sqlx-core", "sqlx-macros-core", - "syn", + "syn 2.0.118", ] [[package]] @@ -2187,7 +2247,7 @@ dependencies = [ "sqlx-mysql", "sqlx-postgres", "sqlx-sqlite", - "syn", + "syn 2.0.118", "tokio", "url", ] @@ -2230,7 +2290,7 @@ dependencies = [ "smallvec", "sqlx-core", "stringprep", - "thiserror", + "thiserror 2.0.18", "tracing", "uuid", "whoami", @@ -2269,7 +2329,7 @@ dependencies = [ "smallvec", "sqlx-core", "stringprep", - "thiserror", + "thiserror 2.0.18", "tracing", "uuid", "whoami", @@ -2295,7 +2355,7 @@ dependencies = [ "serde", "serde_urlencoded", "sqlx-core", - "thiserror", + "thiserror 2.0.18", "tracing", "url", "uuid", @@ -2330,6 +2390,17 @@ version = "2.6.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "13c2bddecc57b384dee18652358fb23172facb8a2c51ccc10d74c157bdea3292" +[[package]] +name = "syn" +version = "1.0.109" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "72b64191b275b66ffe2469e8af2c1cfe3bafa67b529ead792a6d0160888b4237" +dependencies = [ + "proc-macro2", + "quote", + "unicode-ident", +] + [[package]] name = "syn" version = "2.0.118" @@ -2358,7 +2429,7 @@ checksum = "728a70f3dbaf5bab7f0c4b1ac8d7ae5ea60a4b5549c8a5914361c99147a709d2" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -2370,13 +2441,33 @@ dependencies = [ "winapi-util", ] +[[package]] +name = "thiserror" +version = "1.0.69" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b6aaf5339b578ea85b50e080feb250a3e8ae8cfcdff9a461c9ec2904bc923f52" +dependencies = [ + "thiserror-impl 1.0.69", +] + [[package]] name = "thiserror" version = "2.0.18" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "4288b5bcbc7920c07a1149a35cf9590a2aa808e0bc1eafaade0b80947865fbc4" dependencies = [ - "thiserror-impl", + "thiserror-impl 2.0.18", +] + +[[package]] +name = "thiserror-impl" +version = "1.0.69" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4fee6c4efc90059e10f81e6d42c60a18f76588c3d74cb83a0b242a2b6c7504c1" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.118", ] [[package]] @@ -2387,7 +2478,7 @@ checksum = "ebc4ee7f67670e9b64d05fa4253e753e016c6c95ff35b89b7941d6b856dec1d5" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -2449,7 +2540,7 @@ checksum = "385a6cb71ab9ab790c5fe8d67f1645e6c450a7ce006a33de03daa956cf70a496" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -2611,7 +2702,7 @@ checksum = "7490cfa5ec963746568740651ac6781f701c9c5ea257c58e057f3ba8cf69e8da" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -2681,7 +2772,7 @@ dependencies = [ "chrono", "lazy_static", "serde_json", - "thiserror", + "thiserror 2.0.18", "ts-rs-macros", "uuid", ] @@ -2694,7 +2785,7 @@ checksum = "0e9d8656589772eeec2cf7a8264d9cda40fb28b9bc53118ceb9e8c07f8f38730" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", "termcolor", ] @@ -2711,7 +2802,7 @@ dependencies = [ "log", "rand 0.9.4", "sha1", - "thiserror", + "thiserror 2.0.18", ] [[package]] @@ -2889,7 +2980,7 @@ dependencies = [ "bumpalo", "proc-macro2", "quote", - "syn", + "syn 2.0.118", "wasm-bindgen-shared", ] @@ -2980,7 +3071,7 @@ checksum = "053e2e040ab57b9dc951b72c264860db7eb3b0200ba345b4e4c3b14f67855ddf" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -2991,7 +3082,7 @@ checksum = "3f316c4a2570ba26bbec722032c4099d8c8bc095efccdc15688708623367e358" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -3196,6 +3287,12 @@ version = "0.6.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1ffae5123b2d3fc086436f8834ae3ab053a283cfac8fe0a0b8eaae044768a4c4" +[[package]] +name = "xmlparser" +version = "0.13.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "66fee0b777b0f5ac1c69bb06d361268faafa61cd4682ae064a171c16c433e9e4" + [[package]] name = "yansi" version = "1.0.1" @@ -3221,7 +3318,7 @@ checksum = "de844c262c8848816172cef550288e7dc6c7b7814b4ee56b3e1553f275f1858e" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", "synstructure", ] @@ -3242,7 +3339,7 @@ checksum = "4714fd92cf900833d49538023a9b3915155210801d1c1169eba513b2addefd71" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -3262,7 +3359,7 @@ checksum = "11532158c46691caf0f2593ea8358fed6bbf68a0315e80aae9bd41fbade684a1" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", "synstructure", ] @@ -3302,7 +3399,7 @@ checksum = "625dc425cab0dca6dc3c3319506e6593dcb08a9f387ea3b284dbd52a92c40555" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] diff --git a/Cargo.toml b/Cargo.toml index 5db5859..bf3f692 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -14,6 +14,7 @@ authors = ["Rob Thijssen "] newsfeed-entities = { path = "crates/newsfeed-entities", version = "=0.1.0" } newsfeed-core = { path = "crates/newsfeed-core", version = "=0.1.0" } newsfeed-data = { path = "crates/newsfeed-data", version = "=0.1.0" } +newsfeed-fetch = { path = "crates/newsfeed-fetch", version = "=0.1.0" } # async runtime + web tokio = { version = "1", features = ["full"] } @@ -54,9 +55,12 @@ rand = "0.8" sha2 = "0.10" base64 = "0.22" -# worker: feed sourcing +# feed sourcing (worker poll + api discovery, via newsfeed-fetch) reqwest = { version = "0.12", default-features = false, features = ["rustls-tls", "gzip", "json"] } feed-rs = "2" +url = "2" +regex = "1" +opml = "1" # cli niceties clap = { version = "4", features = ["derive", "env"] } diff --git a/crates/newsfeed-api/Cargo.toml b/crates/newsfeed-api/Cargo.toml index dc80036..d209c39 100644 --- a/crates/newsfeed-api/Cargo.toml +++ b/crates/newsfeed-api/Cargo.toml @@ -15,6 +15,7 @@ path = "src/main.rs" newsfeed-entities.workspace = true newsfeed-core.workspace = true newsfeed-data.workspace = true +newsfeed-fetch.workspace = true tokio.workspace = true axum.workspace = true diff --git a/crates/newsfeed-api/src/main.rs b/crates/newsfeed-api/src/main.rs index 02c41f0..3eb98a7 100644 --- a/crates/newsfeed-api/src/main.rs +++ b/crates/newsfeed-api/src/main.rs @@ -47,8 +47,11 @@ async fn main() -> anyhow::Result<()> { .await .context("opening database")?; + let prober = + std::sync::Arc::new(newsfeed_fetch::HttpFeedProber::new().context("building feed prober")?); + let bind = config.bind; - let state = AppState::new(store, config); + let state = AppState::new(store, config, prober); let app = routes::router(state); let listener = TcpListener::bind(bind) diff --git a/crates/newsfeed-api/src/routes/mod.rs b/crates/newsfeed-api/src/routes/mod.rs index 51b06b2..45ea2d1 100644 --- a/crates/newsfeed-api/src/routes/mod.rs +++ b/crates/newsfeed-api/src/routes/mod.rs @@ -33,6 +33,8 @@ pub fn router(state: AppState) -> Router { .route("/feed/signals", post(feed::post_signal)) // sources .route("/sources", get(sources::list).post(sources::create)) + .route("/sources/discover", post(sources::discover)) + .route("/sources/import", post(sources::import)) .route("/sources/{id}", delete(sources::remove)) // interests .route("/interests", get(interests::list).put(interests::upsert)) diff --git a/crates/newsfeed-api/src/routes/sources.rs b/crates/newsfeed-api/src/routes/sources.rs index 9600d34..d46d308 100644 --- a/crates/newsfeed-api/src/routes/sources.rs +++ b/crates/newsfeed-api/src/routes/sources.rs @@ -1,4 +1,6 @@ -//! CRUD for a user's content sources. +//! CRUD for a user's content sources, plus feed discovery and OPML import. + +use std::collections::HashSet; use axum::Json; use axum::extract::{Path, State}; @@ -6,10 +8,13 @@ use axum::http::StatusCode; use uuid::Uuid; use newsfeed_core::ports::SourceStore; -use newsfeed_entities::source::{NewSource, Source}; +use newsfeed_entities::source::{ + DiscoverFeedRequest, DiscoveredFeed, NewSource, OpmlImport, OpmlImportResult, Source, + SourceKind, +}; use crate::auth::CurrentUser; -use crate::error::ApiResult; +use crate::error::{ApiError, ApiResult}; use crate::state::AppState; /// `GET /v1/sources` — the user's sources. @@ -20,17 +25,74 @@ pub async fn list( Ok(Json(app.store.list_sources(user.id).await?)) } -/// `POST /v1/sources` — add a source. +/// `POST /v1/sources` — add a source. For RSS the supplied URL may be any page (homepage, +/// channel, subreddit …); it is resolved to the concrete feed it advertises and the +/// resolved feed URL is what gets stored. pub async fn create( CurrentUser(user): CurrentUser, State(app): State, - Json(new): Json, + Json(mut new): Json, ) -> ApiResult<(StatusCode, Json)> { new.validate()?; + if new.kind == SourceKind::Rss { + let input = new.url.clone().unwrap_or_default(); + let discovered = app.prober.probe(&input).await?; + new.url = Some(discovered.feed_url); + } let source = app.store.create_source(user.id, &new).await?; Ok((StatusCode::CREATED, Json(source))) } +/// `POST /v1/sources/discover` — resolve a page URL to a feed without saving anything, so +/// the UI can preview the feed URL + title before the user commits. +pub async fn discover( + CurrentUser(_user): CurrentUser, + State(app): State, + Json(req): Json, +) -> ApiResult> { + Ok(Json(app.prober.probe(&req.url).await?)) +} + +/// `POST /v1/sources/import` — bulk-add RSS sources from an OPML document. Feeds already +/// present (by URL) or repeated within the file are skipped; per-feed failures are +/// collected rather than aborting the whole import. +pub async fn import( + CurrentUser(user): CurrentUser, + State(app): State, + Json(req): Json, +) -> ApiResult> { + let feeds = newsfeed_fetch::parse_opml(&req.opml) + .map_err(|e| ApiError::bad_request(format!("{e:#}")))?; + + // Dedup against existing sources and against repeats within the file. + let mut seen: HashSet = app + .store + .list_sources(user.id) + .await? + .into_iter() + .filter_map(|s| s.url) + .collect(); + + let mut result = OpmlImportResult::default(); + for feed in feeds { + if !seen.insert(feed.xml_url.clone()) { + result.skipped += 1; + continue; + } + let new = NewSource { + kind: SourceKind::Rss, + name: feed.title.unwrap_or_else(|| feed.xml_url.clone()), + url: Some(feed.xml_url.clone()), + weight: None, + }; + match app.store.create_source(user.id, &new).await { + Ok(_) => result.added += 1, + Err(e) => result.failed.push(format!("{}: {e}", feed.xml_url)), + } + } + Ok(Json(result)) +} + /// `DELETE /v1/sources/{id}` — remove a source. pub async fn remove( CurrentUser(user): CurrentUser, diff --git a/crates/newsfeed-api/src/state.rs b/crates/newsfeed-api/src/state.rs index c54b469..b01fbec 100644 --- a/crates/newsfeed-api/src/state.rs +++ b/crates/newsfeed-api/src/state.rs @@ -3,6 +3,7 @@ use std::sync::Arc; +use newsfeed_core::ports::FeedProbe; use newsfeed_data::SqliteStore; use crate::config::Config; @@ -12,13 +13,16 @@ use crate::config::Config; pub struct AppState { pub store: Arc, pub config: Arc, + /// Resolves page URLs to feeds when a user adds an RSS source. + pub prober: Arc, } impl AppState { - pub fn new(store: SqliteStore, config: Config) -> Self { + pub fn new(store: SqliteStore, config: Config, prober: Arc) -> Self { Self { store: Arc::new(store), config: Arc::new(config), + prober, } } } diff --git a/crates/newsfeed-core/src/ports.rs b/crates/newsfeed-core/src/ports.rs index 186f239..4c5f864 100644 --- a/crates/newsfeed-core/src/ports.rs +++ b/crates/newsfeed-core/src/ports.rs @@ -8,7 +8,7 @@ use uuid::Uuid; use newsfeed_entities::auth::{ApiTokenInfo, Session}; use newsfeed_entities::interest::{Interest, UpsertInterest}; use newsfeed_entities::item::ContentItem; -use newsfeed_entities::source::{NewSource, Source}; +use newsfeed_entities::source::{DiscoveredFeed, NewSource, Source}; use newsfeed_entities::user::User; use crate::error::CoreResult; @@ -96,6 +96,16 @@ pub trait SourceStore: Send + Sync { async fn mark_polled(&self, source_id: Uuid, at: DateTime) -> CoreResult<()>; } +/// Resolves an arbitrary page URL to a concrete feed. Implemented by an outbound-HTTP +/// adapter (`newsfeed-fetch`), which is I/O and therefore lives outside this crate; the +/// port keeps the API handlers backend-agnostic and unit-testable. +#[async_trait] +pub trait FeedProbe: Send + Sync { + /// Resolve `input_url` (a homepage, channel, subreddit, or direct feed URL) to the + /// feed it advertises, verifying the target actually parses as RSS/Atom. + async fn probe(&self, input_url: &str) -> CoreResult; +} + /// User interests / weightings. #[async_trait] pub trait InterestStore: Send + Sync { diff --git a/crates/newsfeed-entities/src/source.rs b/crates/newsfeed-entities/src/source.rs index 87d8354..9ff3997 100644 --- a/crates/newsfeed-entities/src/source.rs +++ b/crates/newsfeed-entities/src/source.rs @@ -73,6 +73,45 @@ pub struct NewSource { pub weight: Option, } +/// Request to resolve an arbitrary page URL to a concrete feed (`POST /v1/sources/discover`). +#[derive(Debug, Clone, Deserialize, TS)] +#[ts(export)] +pub struct DiscoverFeedRequest { + /// Any page URL: a blog homepage, a YouTube channel/video, a subreddit, a Mastodon + /// profile, or an already-direct feed URL. + pub url: String, +} + +/// A feed resolved from a page URL: the concrete feed URL to poll, plus its title. +#[derive(Debug, Clone, Serialize, Deserialize, TS)] +#[ts(export)] +pub struct DiscoveredFeed { + /// The concrete RSS/Atom URL the worker should poll. + pub feed_url: String, + /// The feed's own title, when the parsed feed advertises one. + pub title: Option, +} + +/// Request to bulk-import RSS sources from an OPML document (`POST /v1/sources/import`). +#[derive(Debug, Clone, Deserialize, TS)] +#[ts(export)] +pub struct OpmlImport { + /// The raw OPML XML (e.g. exported from another reader or a subscriptions manager). + pub opml: String, +} + +/// Outcome of an OPML import. +#[derive(Debug, Clone, Default, Serialize, Deserialize, TS)] +#[ts(export)] +pub struct OpmlImportResult { + /// Feeds newly added as sources. + pub added: u32, + /// Feeds skipped because a source with that URL already existed (or repeated in the file). + pub skipped: u32, + /// Feeds that failed to import, each as `": "`. + pub failed: Vec, +} + impl NewSource { /// Validate a create-source request. pub fn validate(&self) -> Result<()> { diff --git a/crates/newsfeed-fetch/Cargo.toml b/crates/newsfeed-fetch/Cargo.toml new file mode 100644 index 0000000..d3bb502 --- /dev/null +++ b/crates/newsfeed-fetch/Cargo.toml @@ -0,0 +1,25 @@ +[package] +name = "newsfeed-fetch" +version.workspace = true +edition.workspace = true +rust-version.workspace = true +license.workspace = true +authors.workspace = true +description = "Outbound feed I/O for newsfeed: RSS/Atom fetch, feed discovery, OPML parse." + +[dependencies] +newsfeed-entities.workspace = true +newsfeed-core.workspace = true + +reqwest.workspace = true +feed-rs.workspace = true +url.workspace = true +regex.workspace = true +opml.workspace = true + +chrono.workspace = true +anyhow.workspace = true +async-trait.workspace = true + +[dev-dependencies] +tokio.workspace = true diff --git a/crates/newsfeed-fetch/src/lib.rs b/crates/newsfeed-fetch/src/lib.rs new file mode 100644 index 0000000..37a381f --- /dev/null +++ b/crates/newsfeed-fetch/src/lib.rs @@ -0,0 +1,374 @@ +//! Outbound feed I/O — the one place that talks HTTP to the wider web. +//! +//! Two consumers: +//! - the **worker** polls known feed URLs via [`fetch_entries`]; +//! - the **API** resolves an arbitrary page URL to a concrete feed via [`HttpFeedProber`] +//! (the [`newsfeed_core::ports::FeedProbe`] adapter) when a user adds a source, and +//! bulk-imports feeds from OPML via [`parse_opml`]. +//! +//! Kept out of `newsfeed-core` (which is I/O-free) and `newsfeed-data` (SQLite only); +//! this is the feed adapter crate. + +use std::sync::OnceLock; +use std::time::Duration; + +use anyhow::Context; +use async_trait::async_trait; +use regex::Regex; +use reqwest::Client; +use url::Url; + +use newsfeed_core::error::{CoreError, CoreResult}; +use newsfeed_core::ingest::FeedEntry; +use newsfeed_core::ports::FeedProbe; +use newsfeed_entities::item::Media; +use newsfeed_entities::source::DiscoveredFeed; + +/// Default User-Agent for outbound feed requests. Some hosts 403 an empty UA. +const DEFAULT_UA: &str = concat!( + "newsfeed-fetch/", + env!("CARGO_PKG_VERSION"), + " (feed discovery)" +); + +// --------------------------------------------------------------------------- +// Worker path: fetch a known feed URL and parse it into backend-agnostic entries. +// --------------------------------------------------------------------------- + +/// Fetch `url` and parse it, returning one [`FeedEntry`] per feed item. +pub async fn fetch_entries(client: &Client, url: &str) -> anyhow::Result> { + let bytes = get_bytes(client, url).await?; + let feed = feed_rs::parser::parse(&bytes[..]).with_context(|| format!("parsing feed {url}"))?; + Ok(feed.entries.into_iter().map(entry_to_feed_entry).collect()) +} + +async fn get_bytes(client: &Client, url: &str) -> anyhow::Result> { + let bytes = client + .get(url) + .send() + .await + .with_context(|| format!("fetching {url}"))? + .error_for_status() + .with_context(|| format!("bad status from {url}"))? + .bytes() + .await + .with_context(|| format!("reading body from {url}"))?; + Ok(bytes.to_vec()) +} + +fn entry_to_feed_entry(entry: feed_rs::model::Entry) -> FeedEntry { + let url = entry.links.first().map(|l| l.href.clone()); + // Prefer the entry's own id; fall back to the link so de-dup still works. + let external_id = if entry.id.is_empty() { + url.clone().unwrap_or_else(|| { + entry + .title + .as_ref() + .map(|t| t.content.clone()) + .unwrap_or_default() + }) + } else { + entry.id.clone() + }; + let title = entry + .title + .map(|t| t.content) + .unwrap_or_else(|| "(untitled)".to_string()); + let summary = entry + .summary + .map(|t| t.content) + .or_else(|| entry.content.and_then(|c| c.body)); + let author = entry.authors.first().map(|p| p.name.clone()); + let tags = entry.categories.into_iter().map(|c| c.term).collect(); + let published_at = entry.published.or(entry.updated); + + let media = entry + .media + .into_iter() + .flat_map(|m| m.content) + .filter_map(|c| { + c.url.map(|u| Media { + kind: c + .content_type + .map(|m| m.to_string()) + .unwrap_or_else(|| "media".to_string()), + url: u.to_string(), + width: c.width, + height: c.height, + }) + }) + .collect(); + + FeedEntry { + external_id, + url, + title, + summary, + author, + tags, + media, + published_at, + } +} + +// --------------------------------------------------------------------------- +// API path: resolve an arbitrary page URL to a concrete feed. +// --------------------------------------------------------------------------- + +/// HTTP adapter that resolves page URLs to feeds. Cheap to clone (shares a `reqwest` +/// connection pool). +#[derive(Clone)] +pub struct HttpFeedProber { + client: Client, +} + +impl HttpFeedProber { + /// Build a prober with a default HTTP client (redirect-following, 20s timeout). + pub fn new() -> anyhow::Result { + let client = Client::builder() + .user_agent(DEFAULT_UA) + .timeout(Duration::from_secs(20)) + .build() + .context("building feed-discovery HTTP client")?; + Ok(Self { client }) + } + + /// Build a prober over a caller-supplied client. + pub fn with_client(client: Client) -> Self { + Self { client } + } + + /// The discovery algorithm, as `anyhow` so callers can map it to their own error type. + async fn discover(&self, input_url: &str) -> anyhow::Result { + let start = normalize_input(input_url).context("that doesn't look like a URL")?; + + // Fetch the (possibly rewritten) target once. If the body itself is a feed we're + // done; otherwise treat it as HTML and look for an advertised feed link. + let (final_url, bytes) = self.fetch(start.as_str()).await?; + + if let Ok(feed) = feed_rs::parser::parse(&bytes[..]) { + return Ok(DiscoveredFeed { + feed_url: final_url.to_string(), + title: feed.title.map(|t| t.content), + }); + } + + let html = String::from_utf8_lossy(&bytes); + let href = find_feed_link(&html) + .with_context(|| format!("no RSS/Atom feed found at {}", final_url))?; + let feed_url = final_url.join(&href).map(|u| u.to_string()).unwrap_or(href); + + // Validate the discovered feed actually parses, and take its title. + let (feed_final, feed_bytes) = self.fetch(&feed_url).await?; + let feed = feed_rs::parser::parse(&feed_bytes[..]) + .with_context(|| format!("discovered feed {feed_url} did not parse"))?; + Ok(DiscoveredFeed { + feed_url: feed_final.to_string(), + title: feed.title.map(|t| t.content), + }) + } + + /// Fetch a URL, returning the final URL (after redirects) and the body bytes. + async fn fetch(&self, url: &str) -> anyhow::Result<(Url, Vec)> { + let resp = self + .client + .get(url) + .send() + .await + .with_context(|| format!("fetching {url}"))? + .error_for_status() + .with_context(|| format!("bad status from {url}"))?; + let final_url = resp.url().clone(); + let bytes = resp + .bytes() + .await + .with_context(|| format!("reading body from {url}"))?; + Ok((final_url, bytes.to_vec())) + } +} + +#[async_trait] +impl FeedProbe for HttpFeedProber { + async fn probe(&self, input_url: &str) -> CoreResult { + // Any failure here is a problem with the URL the user supplied, so surface it as a + // domain-validation error (→ HTTP 400) with the underlying cause chained in. + self.discover(input_url).await.map_err(|e| { + CoreError::Domain(newsfeed_entities::Error::invalid("url", format!("{e:#}"))) + }) + } +} + +/// Normalise a user-typed URL: trim, default to `https://` when no scheme is given, then +/// apply high-confidence feed rewrites (currently Reddit `→ /.rss`). Returns the URL to +/// fetch first; if it isn't a feed the discovery step falls back to HTML link-scanning. +fn normalize_input(input: &str) -> Option { + let trimmed = input.trim(); + if trimmed.is_empty() { + return None; + } + let with_scheme = if trimmed.contains("://") { + trimmed.to_string() + } else { + format!("https://{trimmed}") + }; + let mut url = Url::parse(&with_scheme).ok()?; + reddit_rewrite(&mut url); + Some(url) +} + +/// Reddit listing pages are JS-rendered and don't advertise their feed in HTML, but every +/// listing has a sibling `.rss`. Rewrite `/r/` and `/user/` to their feed. +fn reddit_rewrite(url: &mut Url) { + let host = url.host_str().unwrap_or(""); + if !(host == "reddit.com" || host.ends_with(".reddit.com")) { + return; + } + let path = url.path().trim_end_matches('/'); + if path.ends_with(".rss") { + return; + } + let is_listing = path.starts_with("/r/") || path.starts_with("/user/"); + if is_listing { + url.set_path(&format!("{path}/.rss")); + } +} + +/// Scan HTML for the first `` +/// and return its `href`. Covers YouTube channels, Mastodon profiles, Substack, WordPress, +/// and most blogs, which all advertise their feed this way. +pub fn find_feed_link(html: &str) -> Option { + static LINK_TAG: OnceLock = OnceLock::new(); + static TYPE_ATTR: OnceLock = OnceLock::new(); + static HREF_ATTR: OnceLock = OnceLock::new(); + + let link_tag = LINK_TAG.get_or_init(|| Regex::new(r"(?is)]*>").unwrap()); + let type_attr = TYPE_ATTR + .get_or_init(|| Regex::new(r#"(?is)type\s*=\s*["'][^"']*(?:rss|atom)\+xml"#).unwrap()); + let href_attr = + HREF_ATTR.get_or_init(|| Regex::new(r#"(?is)href\s*=\s*["']([^"']+)["']"#).unwrap()); + + for tag in link_tag.find_iter(html) { + let tag = tag.as_str(); + if type_attr.is_match(tag) { + if let Some(href) = href_attr.captures(tag).and_then(|c| c.get(1)) { + let href = href.as_str().trim(); + if !href.is_empty() { + return Some(href.to_string()); + } + } + } + } + None +} + +// --------------------------------------------------------------------------- +// OPML import. +// --------------------------------------------------------------------------- + +/// A feed extracted from an OPML document. +pub struct OpmlFeed { + /// Display title (`title` attr, falling back to `text`), if any. + pub title: Option, + /// The feed URL (`xmlUrl` attr). + pub xml_url: String, +} + +/// Parse an OPML document into its feeds, recursing through nested outline folders. +pub fn parse_opml(text: &str) -> anyhow::Result> { + let doc = opml::OPML::from_str(text).map_err(|e| anyhow::anyhow!("invalid OPML: {e}"))?; + let mut out = Vec::new(); + collect_outlines(&doc.body.outlines, &mut out); + Ok(out) +} + +fn collect_outlines(outlines: &[opml::Outline], out: &mut Vec) { + for o in outlines { + if let Some(xml_url) = &o.xml_url { + let title = o + .title + .clone() + .filter(|t| !t.is_empty()) + .or_else(|| (!o.text.is_empty()).then(|| o.text.clone())); + out.push(OpmlFeed { + title, + xml_url: xml_url.clone(), + }); + } + collect_outlines(&o.outlines, out); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn reddit_listing_rewrites_to_rss() { + let u = normalize_input("reddit.com/r/rust").unwrap(); + assert_eq!(u.as_str(), "https://reddit.com/r/rust/.rss"); + let u = normalize_input("https://www.reddit.com/r/rust/").unwrap(); + assert_eq!(u.as_str(), "https://www.reddit.com/r/rust/.rss"); + let u = normalize_input("https://reddit.com/user/spez").unwrap(); + assert_eq!(u.as_str(), "https://reddit.com/user/spez/.rss"); + } + + #[test] + fn reddit_existing_rss_untouched() { + let u = normalize_input("https://reddit.com/r/rust/.rss").unwrap(); + assert_eq!(u.as_str(), "https://reddit.com/r/rust/.rss"); + } + + #[test] + fn non_reddit_passes_through_with_scheme() { + let u = normalize_input("example.com/blog").unwrap(); + assert_eq!(u.as_str(), "https://example.com/blog"); + } + + #[test] + fn empty_input_is_none() { + assert!(normalize_input(" ").is_none()); + } + + #[test] + fn finds_youtube_style_feed_link() { + let html = r#" + + + "#; + assert_eq!( + find_feed_link(html).as_deref(), + Some("https://www.youtube.com/feeds/videos.xml?channel_id=UC123") + ); + } + + #[test] + fn finds_atom_link_regardless_of_attr_order() { + let html = r#""#; + assert_eq!(find_feed_link(html).as_deref(), Some("/feed.atom")); + } + + #[test] + fn no_feed_link_returns_none() { + let html = r#""#; + assert!(find_feed_link(html).is_none()); + } + + #[test] + fn parses_opml_with_nested_folders() { + let opml = r#" + subs + + + + + "#; + let feeds = parse_opml(opml).unwrap(); + assert_eq!(feeds.len(), 2); + assert_eq!(feeds[0].xml_url, "https://hnrss.org/frontpage"); + assert_eq!(feeds[0].title.as_deref(), Some("Hacker News")); + assert_eq!(feeds[1].xml_url, "https://example.com/feed.xml"); + assert_eq!(feeds[1].title.as_deref(), Some("Blog")); + } +} diff --git a/crates/newsfeed-worker/Cargo.toml b/crates/newsfeed-worker/Cargo.toml index 688b71b..f5f4375 100644 --- a/crates/newsfeed-worker/Cargo.toml +++ b/crates/newsfeed-worker/Cargo.toml @@ -15,10 +15,10 @@ path = "src/main.rs" newsfeed-entities.workspace = true newsfeed-core.workspace = true newsfeed-data.workspace = true +newsfeed-fetch.workspace = true tokio.workspace = true reqwest.workspace = true -feed-rs.workspace = true serde.workspace = true chrono.workspace = true diff --git a/crates/newsfeed-worker/src/main.rs b/crates/newsfeed-worker/src/main.rs index 3917405..9853936 100644 --- a/crates/newsfeed-worker/src/main.rs +++ b/crates/newsfeed-worker/src/main.rs @@ -6,7 +6,6 @@ //! re-seeing an item's `external_id` is a no-op. mod config; -mod sourcing; use std::collections::HashMap; use std::path::PathBuf; @@ -25,7 +24,6 @@ use newsfeed_data::SqliteStore; use newsfeed_entities::interest::Interest; use crate::config::Config; -use crate::sourcing::rss; #[derive(Parser)] #[command( @@ -90,7 +88,7 @@ async fn run_cycle(store: &SqliteStore, client: &Client, cfg: &Config) -> anyhow continue; }; - match rss::fetch(client, &url).await { + match newsfeed_fetch::fetch_entries(client, &url).await { Ok(entries) => { let interests = if let Some(cached) = interests_cache.get(&source.user_id) { cached diff --git a/crates/newsfeed-worker/src/sourcing/mod.rs b/crates/newsfeed-worker/src/sourcing/mod.rs deleted file mode 100644 index 40972b1..0000000 --- a/crates/newsfeed-worker/src/sourcing/mod.rs +++ /dev/null @@ -1,3 +0,0 @@ -//! Content sourcing. Currently RSS/Atom polling; other algorithmic sources plug in here. - -pub mod rss; diff --git a/crates/newsfeed-worker/src/sourcing/rss.rs b/crates/newsfeed-worker/src/sourcing/rss.rs deleted file mode 100644 index a262cc7..0000000 --- a/crates/newsfeed-worker/src/sourcing/rss.rs +++ /dev/null @@ -1,80 +0,0 @@ -//! Fetch and parse an RSS/Atom feed into backend-agnostic [`FeedEntry`] values. - -use anyhow::Context; -use reqwest::Client; - -use newsfeed_core::ingest::FeedEntry; -use newsfeed_entities::item::Media; - -/// Fetch `url` and parse it, returning one [`FeedEntry`] per feed item. -pub async fn fetch(client: &Client, url: &str) -> anyhow::Result> { - let bytes = client - .get(url) - .send() - .await - .with_context(|| format!("fetching {url}"))? - .error_for_status() - .with_context(|| format!("bad status from {url}"))? - .bytes() - .await - .with_context(|| format!("reading body from {url}"))?; - - let feed = feed_rs::parser::parse(&bytes[..]).with_context(|| format!("parsing feed {url}"))?; - - Ok(feed.entries.into_iter().map(entry_to_feed_entry).collect()) -} - -fn entry_to_feed_entry(entry: feed_rs::model::Entry) -> FeedEntry { - let url = entry.links.first().map(|l| l.href.clone()); - // Prefer the entry's own id; fall back to the link so de-dup still works. - let external_id = if entry.id.is_empty() { - url.clone().unwrap_or_else(|| { - entry - .title - .as_ref() - .map(|t| t.content.clone()) - .unwrap_or_default() - }) - } else { - entry.id.clone() - }; - let title = entry - .title - .map(|t| t.content) - .unwrap_or_else(|| "(untitled)".to_string()); - let summary = entry - .summary - .map(|t| t.content) - .or_else(|| entry.content.and_then(|c| c.body)); - let author = entry.authors.first().map(|p| p.name.clone()); - let tags = entry.categories.into_iter().map(|c| c.term).collect(); - let published_at = entry.published.or(entry.updated); - - let media = entry - .media - .into_iter() - .flat_map(|m| m.content) - .filter_map(|c| { - c.url.map(|u| Media { - kind: c - .content_type - .map(|m| m.to_string()) - .unwrap_or_else(|| "media".to_string()), - url: u.to_string(), - width: c.width, - height: c.height, - }) - }) - .collect(); - - FeedEntry { - external_id, - url, - title, - summary, - author, - tags, - media, - published_at, - } -} diff --git a/web/src/api/bindings/DiscoverFeedRequest.ts b/web/src/api/bindings/DiscoverFeedRequest.ts new file mode 100644 index 0000000..f5a7118 --- /dev/null +++ b/web/src/api/bindings/DiscoverFeedRequest.ts @@ -0,0 +1,11 @@ +// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually. + +/** + * Request to resolve an arbitrary page URL to a concrete feed (`POST /v1/sources/discover`). + */ +export type DiscoverFeedRequest = { +/** + * Any page URL: a blog homepage, a YouTube channel/video, a subreddit, a Mastodon + * profile, or an already-direct feed URL. + */ +url: string, }; diff --git a/web/src/api/bindings/DiscoveredFeed.ts b/web/src/api/bindings/DiscoveredFeed.ts new file mode 100644 index 0000000..443939f --- /dev/null +++ b/web/src/api/bindings/DiscoveredFeed.ts @@ -0,0 +1,14 @@ +// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually. + +/** + * A feed resolved from a page URL: the concrete feed URL to poll, plus its title. + */ +export type DiscoveredFeed = { +/** + * The concrete RSS/Atom URL the worker should poll. + */ +feed_url: string, +/** + * The feed's own title, when the parsed feed advertises one. + */ +title: string | null, }; diff --git a/web/src/api/bindings/OpmlImport.ts b/web/src/api/bindings/OpmlImport.ts new file mode 100644 index 0000000..90ee048 --- /dev/null +++ b/web/src/api/bindings/OpmlImport.ts @@ -0,0 +1,10 @@ +// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually. + +/** + * Request to bulk-import RSS sources from an OPML document (`POST /v1/sources/import`). + */ +export type OpmlImport = { +/** + * The raw OPML XML (e.g. exported from another reader or a subscriptions manager). + */ +opml: string, }; diff --git a/web/src/api/bindings/OpmlImportResult.ts b/web/src/api/bindings/OpmlImportResult.ts new file mode 100644 index 0000000..56ff022 --- /dev/null +++ b/web/src/api/bindings/OpmlImportResult.ts @@ -0,0 +1,18 @@ +// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually. + +/** + * Outcome of an OPML import. + */ +export type OpmlImportResult = { +/** + * Feeds newly added as sources. + */ +added: number, +/** + * Feeds skipped because a source with that URL already existed (or repeated in the file). + */ +skipped: number, +/** + * Feeds that failed to import, each as `": "`. + */ +failed: Array, }; diff --git a/web/src/api/client.ts b/web/src/api/client.ts index 68cd776..df7064e 100644 --- a/web/src/api/client.ts +++ b/web/src/api/client.ts @@ -10,6 +10,8 @@ import type { CreateApiToken } from './bindings/CreateApiToken'; import type { CreatedApiToken } from './bindings/CreatedApiToken'; import type { Source } from './bindings/Source'; import type { NewSource } from './bindings/NewSource'; +import type { DiscoveredFeed } from './bindings/DiscoveredFeed'; +import type { OpmlImportResult } from './bindings/OpmlImportResult'; import type { Interest } from './bindings/Interest'; import type { UpsertInterest } from './bindings/UpsertInterest'; import type { FeedPage } from './bindings/FeedPage'; @@ -71,6 +73,8 @@ export const api = { listSources: () => request('GET', '/v1/sources'), createSource: (s: NewSource) => request('POST', '/v1/sources', s), deleteSource: (id: string) => request('DELETE', `/v1/sources/${id}`), + discoverFeed: (url: string) => request('POST', '/v1/sources/discover', { url }), + importOpml: (opml: string) => request('POST', '/v1/sources/import', { opml }), // interests listInterests: () => request('GET', '/v1/interests'), diff --git a/web/src/routes/Sources.tsx b/web/src/routes/Sources.tsx index 78bdec2..a3863ea 100644 --- a/web/src/routes/Sources.tsx +++ b/web/src/routes/Sources.tsx @@ -1,9 +1,19 @@ -import { useState, type FormEvent } from 'react'; +import { useRef, useState, type ChangeEvent, type FormEvent } from 'react'; import { useMutation, useQuery, useQueryClient } from '@tanstack/react-query'; import { api, ApiError } from '../api/client'; import type { SourceKind } from '../api/bindings/SourceKind'; +/** Human "polled 5m ago" hint so the user can see the RSS rail actually working. */ +function pollStatus(iso: string | null): string { + if (!iso) return 'never polled'; + const secs = (Date.now() - new Date(iso).getTime()) / 1000; + if (secs < 60) return 'polled just now'; + if (secs < 3600) return `polled ${Math.floor(secs / 60)}m ago`; + if (secs < 86400) return `polled ${Math.floor(secs / 3600)}h ago`; + return `polled ${Math.floor(secs / 86400)}d ago`; +} + export function Sources() { const qc = useQueryClient(); const sources = useQuery({ queryKey: ['sources'], queryFn: api.listSources }); @@ -12,28 +22,67 @@ export function Sources() { const [name, setName] = useState(''); const [url, setUrl] = useState(''); const [weight, setWeight] = useState(1); + const fileInput = useRef(null); + + // Preview of the feed a pasted URL resolves to (populated by "Find feed"). + const discover = useMutation({ + mutationFn: () => api.discoverFeed(url), + onSuccess: (d) => { + if (d.title && !name.trim()) setName(d.title); + }, + }); const create = useMutation({ - mutationFn: () => api.createSource({ kind, name, url: kind === 'rss' ? url : null, weight }), + // Send the resolved feed URL when we have one (fast path); otherwise the raw URL, + // which the server resolves itself. Agentic sources carry no URL. + mutationFn: () => + api.createSource({ + kind, + name, + url: kind === 'rss' ? (discover.data?.feed_url ?? url) : null, + weight, + }), onSuccess: () => { setName(''); setUrl(''); setWeight(1); + discover.reset(); qc.invalidateQueries({ queryKey: ['sources'] }); }, }); + const importOpml = useMutation({ + mutationFn: (opml: string) => api.importOpml(opml), + onSuccess: () => qc.invalidateQueries({ queryKey: ['sources'] }), + }); + const remove = useMutation({ mutationFn: (id: string) => api.deleteSource(id), onSuccess: () => qc.invalidateQueries({ queryKey: ['sources'] }), }); + // Editing the URL invalidates a previous discovery preview. + const onUrlChange = (e: ChangeEvent) => { + setUrl(e.target.value); + if (discover.data || discover.error) discover.reset(); + }; + + const onFile = async (e: ChangeEvent) => { + const file = e.target.files?.[0]; + e.target.value = ''; // allow re-importing the same file + if (file) importOpml.mutate(await file.text()); + }; + const onSubmit = (e: FormEvent) => { e.preventDefault(); create.mutate(); }; - const error = create.error instanceof ApiError ? create.error.message : null; + const errOf = (e: unknown) => (e instanceof ApiError ? e.message : e ? String(e) : null); + const createError = errOf(create.error); + const discoverError = errOf(discover.error); + const importError = errOf(importOpml.error); + const summary = importOpml.data; return (
@@ -57,16 +106,37 @@ export function Sources() { {kind === 'rss' && ( - + <> + + + {discover.data && ( +

+ Found feed{discover.data.title ? ` “${discover.data.title}”` : ''}:{' '} + {discover.data.feed_url} +

+ )} + {discoverError &&

{discoverError}

} + )} - {error &&

{error}

} + {createError &&

{createError}

} +
+
+
+ Import OPML +

+ Bulk-add feeds from a reader export or a YouTube subscriptions OPML. Duplicates are skipped. +

+
+ + +
+ {summary && ( +

+ Imported {summary.added} added, {summary.skipped} skipped + {summary.failed.length > 0 ? `, ${summary.failed.length} failed` : ''}. +

+ )} + {summary && summary.failed.length > 0 && ( +
    + {summary.failed.map((f) => ( +
  • {f}
    • + ))} +
+ )} + {importError &&

{importError}

} +
+
    {sources.data?.map((s) => (
  • @@ -100,6 +210,7 @@ export function Sources() { )}
    + {s.kind === 'rss' && {pollStatus(s.last_polled_at)}} w {s.weight.toFixed(2)}