feat: synchronize source coverage

Cargo.lock

@@ -136,7 +136,7 @@ dependencies = [
 
 [[package]]
 name = "basango"
-version = "0.1.3"
+version = "0.2.0"
 dependencies = [
  "anyhow",
  "bullmq-official",

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "basango"
-version = "0.1.3"
+version = "0.2.0"
 edition = "2024"
 rust-version = "1.85"
 description = "A Rust-native Basango news crawler with HTML and WordPress adapters"

config/crawler.json

@@ -21,7 +21,7 @@
     "max_retries": 3,
     "rotate": true,
     "timeout": 20,
-    "user_agent": "Basango/0.1 (+https://basango.ngandu.dev)"
+    "user_agent": "Basango/0.2 (+https://basango.ngandu.dev)"
   },
   "runtime": {
     "direction": "forward",

src/config/http.rs

@@ -30,7 +30,7 @@ impl Default for HttpClientConfig {
             respect_retry_after: true,
             rotate: true,
             timeout: 20,
-            user_agent: "Basango/0.1 (+https://github.com/bernard-ng/basango)".into(),
+            user_agent: "Basango/0.2 (+https://github.com/bernard-ng/basango)".into(),
             verify_ssl: true,
         }
     }

src/config/source.rs

@@ -89,6 +89,13 @@ impl SourceConfig {
         &self.common().url
     }
 
+    pub fn kind(&self) -> &'static str {
+        match self {
+            Self::Html(_) => "html",
+            Self::WordPress(_) => "wordpress",
+        }
+    }
+
     pub fn common(&self) -> &CommonSourceConfig {
         match self {
             Self::Html(source) => &source.common,

src/crawler.rs

@@ -115,11 +115,13 @@ impl Crawler {
 
     /// Crawl now, streaming collected drafts into the durable outbox.
     pub async fn crawl(&self, request: CrawlRequest) -> Result<CrawlReport> {
+        self.runtime.synchronize_sources().await?;
         crawl_now(&self.runtime, request).await
     }
 
     /// Schedule source discovery in BullMQ.
     pub async fn schedule(&self, mut request: CrawlRequest) -> Result<String> {
+        self.runtime.synchronize_sources().await?;
         self.runtime.config.prepare_request(&mut request)?;
         let reporter = RunReporter::new(
             &self.runtime.config.ingestion,
@@ -173,6 +175,7 @@ impl Crawler {
 
     /// Run BullMQ consumers until the process receives Ctrl-C.
     pub async fn work(&self, queues: Vec<String>, concurrency: usize) -> Result<()> {
+        self.runtime.synchronize_sources().await?;
         run_worker(self.runtime.clone(), queues, concurrency).await
     }
 

src/execution.rs

@@ -4,6 +4,7 @@
 //! deliver. Execution modules coordinate those capabilities for each command.
 
 mod queue;
+mod source_sync;
 mod sync;
 mod worker;
 
@@ -19,6 +20,7 @@ use std::sync::Arc;
 
 use chrono::{DateTime, Utc};
 use serde::Deserialize;
+use tokio::sync::OnceCell;
 
 use crate::{
     articles::endpoint_url,
@@ -36,6 +38,7 @@ pub(crate) struct Runtime {
     pub config: Arc<CrawlerConfig>,
     pub http: HttpClient,
     pub agent_id: String,
+    source_sync: Arc<OnceCell<()>>,
 }
 
 impl Runtime {
@@ -47,9 +50,17 @@ impl Runtime {
             config: Arc::new(config),
             http,
             agent_id,
+            source_sync: Arc::new(OnceCell::new()),
         })
     }
 
+    pub async fn synchronize_sources(&self) -> Result<()> {
+        self.source_sync
+            .get_or_try_init(|| async { source_sync::synchronize(self).await })
+            .await
+            .map(|_| ())
+    }
+
     /// Ask the ingestion API for the last known article boundary when the caller did
     /// not explicitly provide a date range. API unavailability should not
     /// prevent a manual crawl, so failures are logged and treated as no range.

src/execution/source_sync.rs

@@ -0,0 +1,131 @@
+//! Registration and archive-size synchronization for configured sources.
+
+use std::sync::Arc;
+
+use serde::Serialize;
+use tokio::{sync::Semaphore, task::JoinSet};
+
+use crate::{
+    articles::endpoint_url, config::SourceConfig, error::Result, execution::Runtime,
+    sources::SourceAdapter,
+};
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
+#[serde(rename_all = "camelCase")]
+struct SourceSyncItem {
+    #[serde(skip_serializing_if = "Option::is_none")]
+    estimated_articles: Option<usize>,
+    kind: String,
+    name: String,
+    url: String,
+}
+
+#[derive(Serialize)]
+struct SourceSyncPayload<'a> {
+    sources: &'a [SourceSyncItem],
+}
+
+pub(super) async fn synchronize(runtime: &Runtime) -> Result<()> {
+    let Some(base) = &runtime.config.ingestion.endpoint else {
+        tracing::debug!("ingestion API is disabled; skipping source synchronization");
+        return Ok(());
+    };
+    let endpoint = endpoint_url(base, "ingest/sources/sync")?;
+    let registrations = runtime
+        .config
+        .sources
+        .iter()
+        .map(SourceSyncItem::from)
+        .collect::<Vec<_>>();
+
+    publish(runtime, &endpoint, &registrations).await?;
+    tracing::info!(
+        sources = registrations.len(),
+        "registered configured crawler sources"
+    );
+
+    let estimates = estimate_sources(runtime).await;
+    if estimates.is_empty() {
+        tracing::warn!("no source archive estimates were available to synchronize");
+        return Ok(());
+    }
+
+    publish(runtime, &endpoint, &estimates).await?;
+    tracing::info!(
+        estimated = estimates.len(),
+        total = registrations.len(),
+        "synchronized source archive estimates"
+    );
+
+    Ok(())
+}
+
+async fn estimate_sources(runtime: &Runtime) -> Vec<SourceSyncItem> {
+    let permits = Arc::new(Semaphore::new(
+        runtime.config.runtime.worker_concurrency.max(1),
+    ));
+    let mut tasks = JoinSet::new();
+
+    for source in runtime.config.sources.iter().cloned() {
+        let http = runtime.http.clone();
+        let permits = permits.clone();
+
+        tasks.spawn(async move {
+            let _permit = permits
+                .acquire_owned()
+                .await
+                .expect("source estimate semaphore remains open");
+            let registration = SourceSyncItem::from(&source);
+            let estimate = SourceAdapter::new(source, http)
+                .estimate_total_articles()
+                .await;
+
+            (registration, estimate)
+        });
+    }
+
+    let mut estimates = Vec::new();
+    while let Some(result) = tasks.join_next().await {
+        match result {
+            Ok((mut registration, Ok(estimate))) => {
+                registration.estimated_articles = Some(estimate);
+                estimates.push(registration);
+            }
+            Ok((registration, Err(error))) => tracing::warn!(
+                source = registration.name,
+                %error,
+                "could not estimate source archive size"
+            ),
+            Err(error) => tracing::warn!(%error, "source archive estimate task failed"),
+        }
+    }
+    estimates.sort_unstable_by(|left, right| left.name.cmp(&right.name));
+
+    estimates
+}
+
+async fn publish(runtime: &Runtime, endpoint: &url::Url, sources: &[SourceSyncItem]) -> Result<()> {
+    let headers = [("Authorization", runtime.config.ingestion.token.as_str())];
+    runtime
+        .http
+        .post_json(endpoint, &headers, &SourceSyncPayload { sources })
+        .await?
+        .require_success()?;
+
+    Ok(())
+}
+
+impl From<&SourceConfig> for SourceSyncItem {
+    fn from(source: &SourceConfig) -> Self {
+        Self {
+            estimated_articles: None,
+            kind: source.kind().to_owned(),
+            name: source.id().to_string(),
+            url: source.url().to_string(),
+        }
+    }
+}
+
+#[cfg(test)]
+#[path = "../../tests/unit/execution/source_sync.rs"]
+mod tests;

src/sources.rs

@@ -95,4 +95,11 @@ impl SourceAdapter {
             Self::WordPress(crawler) => crawler.collect(seed, request).await,
         }
     }
+
+    pub async fn estimate_total_articles(&self) -> Result<usize> {
+        match self {
+            Self::Html(crawler) => crawler.estimate_total_articles().await,
+            Self::WordPress(crawler) => crawler.estimate_total_articles().await,
+        }
+    }
 }

src/sources/html.rs

@@ -30,6 +30,27 @@ impl HtmlCrawler {
         Self { source, http }
     }
 
+    pub async fn estimate_total_articles(&self) -> Result<usize> {
+        let mut estimate = 0usize;
+
+        if self.source.indexed_categories.is_empty() {
+            return self.estimate_category(None).await;
+        }
+
+        for category in &self.source.indexed_categories {
+            estimate = estimate
+                .checked_add(self.estimate_category(Some(category)).await?)
+                .ok_or_else(|| {
+                    CrawlError::Configuration(format!(
+                        "archive estimate overflowed for source '{}'",
+                        self.source.common.id
+                    ))
+                })?;
+        }
+
+        Ok(estimate)
+    }
+
     /// Crawl listings and detail pages directly in one process.
     pub async fn crawl_into(
         &self,
@@ -271,6 +292,37 @@ impl HtmlCrawler {
         let Ok(html) = self.fetch_text(&url).await else {
             return Ok(fallback);
         };
+
+        self.pagination_from_html(&html)
+    }
+
+    async fn estimate_category(&self, category: Option<&str>) -> Result<usize> {
+        let url = self.endpoint_url(0, category)?;
+        let html = self.fetch_text(&url).await?;
+        let page_range = self.pagination_from_html(&html)?;
+        let articles_per_page = self.listing_entries(&html)?.len();
+        if articles_per_page == 0 {
+            return Err(CrawlError::InvalidSourceSelectors(format!(
+                "selector '{}' matched no archive articles for source '{}'",
+                self.source.selectors.list, self.source.common.id
+            )));
+        }
+        let estimate = estimate_archive_size(articles_per_page, page_range)?;
+
+        tracing::info!(
+            source = %self.source.common.id,
+            category = category.unwrap_or("<none>"),
+            articles_per_page,
+            pages = %page_range,
+            estimate,
+            "estimated HTML archive size"
+        );
+
+        Ok(estimate)
+    }
+
+    fn pagination_from_html(&self, html: &str) -> Result<PageRange> {
+        let fallback = PageRange::new(0, 0)?;
         let document = Html::parse_document(&html);
         let selector = parse_selector(&self.source.selectors.pagination)?;
         let Some(href) = document
@@ -397,6 +449,17 @@ impl HtmlCrawler {
     }
 }
 
+fn estimate_archive_size(articles_per_page: usize, page_range: PageRange) -> Result<usize> {
+    let page_count = u64::from(page_range.end) - u64::from(page_range.start) + 1;
+    let page_count = usize::try_from(page_count).map_err(|_| {
+        CrawlError::Configuration("HTML archive page count does not fit this platform".into())
+    })?;
+
+    articles_per_page
+        .checked_mul(page_count)
+        .ok_or_else(|| CrawlError::Configuration("HTML archive article estimate overflowed".into()))
+}
+
 #[cfg(test)]
 #[path = "../../tests/unit/sources/html.rs"]
 mod tests;

src/sources/wordpress.rs

@@ -75,6 +75,12 @@ impl WordPressCrawler {
         }
     }
 
+    pub async fn estimate_total_articles(&self) -> Result<usize> {
+        let (_, articles) = self.archive_summary().await?;
+
+        Ok(articles)
+    }
+
     pub async fn crawl_into(
         &mut self,
         request: &CrawlRequest,
@@ -237,20 +243,31 @@ impl WordPressCrawler {
     }
 
     async fn pagination(&self) -> Result<PageRange> {
+        let (page_range, _) = self.archive_summary().await?;
+
+        Ok(page_range)
+    }
+
+    async fn archive_summary(&self) -> Result<(PageRange, usize)> {
         let mut url = self.api_url("wp-json/wp/v2/posts")?;
         url.query_pairs_mut()
             .append_pair("_fields", "id")
             .append_pair("per_page", "100");
-        let response = self.fetch(&url).await?;
+        let response = self.fetch(&url).await?.require_success()?;
         let pages = header_number(&response.headers, "x-wp-totalpages").unwrap_or(1);
-        let posts = header_number(&response.headers, "x-wp-total").unwrap_or(0);
+        let page_range = PageRange::new(1, pages.max(1))?;
+        let posts = match header_number(&response.headers, "x-wp-total") {
+            Some(total) => total as usize,
+            None => response.json::<Vec<serde_json::Value>>()?.len() * pages.max(1) as usize,
+        };
         tracing::info!(
             pages,
             posts,
             source = %self.source.common.id,
             "WordPress pagination"
         );
-        PageRange::new(1, pages.max(1))
+
+        Ok((page_range, posts))
     }
 
     fn page_url(&self, page: u32) -> Result<Url> {

tests/unit/execution/source_sync.rs

@@ -0,0 +1,37 @@
+use crate::config::{CommonSourceConfig, MetadataStrategy, SourceConfig, WordPressSourceConfig};
+
+use super::*;
+
+#[test]
+fn registration_uses_the_crawler_source_identity() {
+    let source = SourceConfig::WordPress(WordPressSourceConfig {
+        common: CommonSourceConfig {
+            id: crate::domain::SourceId::new("example.com").unwrap(),
+            url: url::Url::parse("https://example.com").unwrap(),
+            ..CommonSourceConfig::default()
+        },
+        metadata_strategy: MetadataStrategy::default(),
+    });
+
+    let registration = SourceSyncItem::from(&source);
+
+    assert_eq!(registration.name, "example.com");
+    assert_eq!(registration.kind, "wordpress");
+    assert_eq!(registration.url, "https://example.com/");
+    assert_eq!(registration.estimated_articles, None);
+}
+
+#[test]
+fn synchronization_payload_omits_an_unavailable_estimate() {
+    let items = vec![SourceSyncItem {
+        estimated_articles: None,
+        kind: "html".into(),
+        name: "example.com".into(),
+        url: "https://example.com".into(),
+    }];
+
+    let value = serde_json::to_value(SourceSyncPayload { sources: &items }).unwrap();
+
+    assert_eq!(value["sources"][0]["name"], "example.com");
+    assert!(value["sources"][0].get("estimatedArticles").is_none());
+}

tests/unit/sources/html.rs

@@ -75,3 +75,34 @@ fn substitutes_category_and_page_in_endpoint() {
         "https://example.com/category/news/page/3"
     );
 }
+
+#[test]
+fn estimates_html_archives_from_first_page_density_and_page_count() {
+    let crawler = HtmlCrawler::new(source(), HttpClient::new(&Default::default()).unwrap());
+    let html = r#"
+        <div class="article"></div>
+        <div class="article"></div>
+        <div class="article"></div>
+        <nav class="pages"><a href="/news?page=4">Last</a></nav>
+    "#;
+
+    let page_range = crawler.pagination_from_html(html).unwrap();
+    let articles_per_page = crawler.listing_entries(html).unwrap().len();
+
+    assert_eq!(page_range, PageRange::new(0, 4).unwrap());
+    assert_eq!(
+        estimate_archive_size(articles_per_page, page_range).unwrap(),
+        15
+    );
+}
+
+#[test]
+fn estimates_a_listing_without_pagination_as_one_page() {
+    let crawler = HtmlCrawler::new(source(), HttpClient::new(&Default::default()).unwrap());
+    let html = r#"<div class="article"></div><div class="article"></div>"#;
+
+    let page_range = crawler.pagination_from_html(html).unwrap();
+
+    assert_eq!(page_range, PageRange::new(0, 0).unwrap());
+    assert_eq!(estimate_archive_size(2, page_range).unwrap(), 2);
+}

tests/unit/sources/wordpress.rs

@@ -1,5 +1,7 @@
 use super::*;
 
+use reqwest::header::{HeaderMap, HeaderValue};
+
 #[test]
 fn extracts_yoast_metadata() {
     let link = Url::parse("https://example.com/story").unwrap();
@@ -28,3 +30,13 @@ fn parses_wordpress_naive_datetime_as_utc() {
         common::parse_published_at("2025-01-02T03:04:05", "yyyy-LL-dd'T'HH:mm:ss").unwrap();
     assert_eq!(date.to_rfc3339(), "2025-01-02T03:04:05+00:00");
 }
+
+#[test]
+fn reads_wordpress_archive_totals_from_headers() {
+    let mut headers = HeaderMap::new();
+    headers.insert("x-wp-total", HeaderValue::from_static("421"));
+    headers.insert("x-wp-totalpages", HeaderValue::from_static("5"));
+
+    assert_eq!(header_number(&headers, "x-wp-total"), Some(421));
+    assert_eq!(header_number(&headers, "x-wp-totalpages"), Some(5));
+}