Initial Commit

This commit is contained in:
2026-08-23 20:15:48 +02:00
commit a7b97db6be
49 changed files with 8283 additions and 0 deletions
+34
View File
@@ -0,0 +1,34 @@
# Ingestion API. Leave the endpoint empty to crawl into SQLite without
# forwarding; the `deliver` command requires an endpoint.
BASANGO_API_CRAWLER_ENDPOINT=
BASANGO_API_CRAWLER_TOKEN=
# Stable agent name shown in the realtime operations dashboard. Defaults to the host name.
BASANGO_CRAWLER_AGENT_ID=
# Optional configuration override. Leave empty to use the JSON embedded in the
# binary at compile time.
BASANGO_CRAWLER_CONFIG_PATH=
# Durable local storage.
BASANGO_CRAWLER_DATA_PATH=data
BASANGO_CRAWLER_SQLITE_PATH=data/crawler.db
# HTTP policy.
BASANGO_CRAWLER_FETCH_USER_AGENT=Basango/0.1 (+https://basango.ngandu.dev)
BASANGO_CRAWLER_FETCH_MAX_RETRIES=3
BASANGO_CRAWLER_FETCH_RESPECT_RETRY_AFTER=true
BASANGO_CRAWLER_UPDATE_DIRECTION=forward
# BullMQ-backed queued execution.
BASANGO_CRAWLER_REDIS_URL=redis://localhost:6379/0
BASANGO_CRAWLER_QUEUE_DISCOVERY=discovery
BASANGO_CRAWLER_QUEUE_ARTICLES=articles
BASANGO_CRAWLER_RETAIN_COMPLETED=3600
BASANGO_CRAWLER_RETAIN_FAILED=86400
# A scheduler can use a different source shard on every machine.
BASANGO_CRAWLER_SOURCE_IDS=radiookapi.net,7sur7.cd
# Standard tracing filter: trace, debug, info, warn, or error.
RUST_LOG=info
+71
View File
@@ -0,0 +1,71 @@
name: CI
on:
push:
pull_request:
permissions:
contents: read
concurrency:
group: ci-${{ github.workflow }}-${{ github.ref }}
cancel-in-progress: true
jobs:
format:
name: Format
runs-on: ubuntu-latest
steps:
- name: Check out repository
uses: actions/checkout@v4
- name: Install Rust
uses: dtolnay/rust-toolchain@stable
with:
components: rustfmt
- name: Check formatting
run: cargo fmt --all -- --check
test:
name: Test
runs-on: ubuntu-latest
steps:
- name: Check out repository
uses: actions/checkout@v4
- name: Install Rust
uses: dtolnay/rust-toolchain@stable
- name: Run tests
run: cargo test --all-targets --locked
build:
name: Build release artifact
if: github.event_name == 'push' && github.ref == 'refs/heads/main'
needs:
- format
- test
runs-on: ubuntu-latest
steps:
- name: Check out repository
uses: actions/checkout@v4
- name: Install Rust
uses: dtolnay/rust-toolchain@stable
- name: Build release binary
run: cargo build --release --locked --bin crawler
- name: Package executable
run: |
mkdir -p dist
tar -C target/release -czf dist/crawler-linux-x86_64.tar.gz crawler
- name: Upload release artifact
uses: actions/upload-artifact@v4
with:
name: crawler-linux-x86_64
path: dist/crawler-linux-x86_64.tar.gz
if-no-files-found: error
retention-days: 7
+9
View File
@@ -0,0 +1,9 @@
/target
.env
.env.local
.idea/
data/
node_modules/
*.db
*.db-shm
*.db-wal
Generated
+3104
View File
File diff suppressed because it is too large Load Diff
+38
View File
@@ -0,0 +1,38 @@
[package]
name = "basango"
version = "0.1.0"
edition = "2024"
rust-version = "1.85"
description = "A Rust-native Basango news crawler with HTML and WordPress adapters"
default-run = "crawler"
[[bin]]
name = "crawler"
path = "src/main.rs"
[dependencies]
anyhow = "1.0.104"
bullmq = { version = "1.2.5", package = "bullmq-official" }
chrono = { version = "0.4.45", features = ["serde"] }
clap = { version = "4.6.6", features = ["derive"] }
dotenvy = "0.15.7"
html2md = "0.2.15"
httpdate = "1.0.3"
md5 = "0.8.0"
rand = "0.9.2"
regex = "1.11.3"
reqwest = { version = "0.12.28", default-features = false, features = ["json", "rustls-tls"] }
rusqlite = { version = "0.37.0", features = ["bundled"] }
scraper = "0.24.0"
serde = { version = "1.0", features = ["derive"] }
serde_json = "1.0"
thiserror = "2.0.17"
tokio = { version = "1.48.0", features = ["macros", "rt-multi-thread", "signal", "sync", "time"] }
tracing = "0.1.41"
tracing-subscriber = { version = "0.3.20", features = ["env-filter", "fmt"] }
url = { version = "2.5", features = ["serde"] }
uuid = { version = "1.25.0", features = ["v7", "serde"] }
zod-rs = { version = "1.0.1", default-features = false }
[dev-dependencies]
tempfile = "3.23.0"
+238
View File
@@ -0,0 +1,238 @@
# Basango Crawler
The Rust crawler is Basango's production collector for Congolese news. It replaces the former TypeScript crawler and runs as an independent service beside the Basango API.
It discovers articles from configured HTML and WordPress sources, normalizes them, saves them to a durable local outbox, and forwards them to the canonical ingestion API. Agent heartbeats and run signals feed Basango's realtime ingestion operations dashboard.
## Features
- **Rust-native pipeline**: typed configuration, requests, articles, errors, and queue payloads.
- **HTML and WordPress sources**: CSS-selector adapters and WordPress REST collection.
- **Direct and queued execution**: immediate crawls plus BullMQ scheduling and concurrent workers.
- **Durable delivery**: SQLite outbox with atomic claims, retryable failures, and idempotency keys.
- **Resilient HTTP**: timeouts, exponential backoff, jitter, `Retry-After`, redirects, and user-agent rotation.
- **Flexible collection windows**: source, page, date, category, and forward/backward update filters.
- **Operational visibility**: idempotent heartbeats and lifecycle signals for Basango's ingestion dashboard.
- **Graceful workers**: shared concurrency limits, BullMQ stalled-job recovery, and clean shutdown.
## Architecture
```text
HTML / WordPress sources
discovery + collection
normalize and validate
SQLite outbox ───────POST /ingest/articles──────► Basango API
│ │
└──────────── retry until delivered ────────────┘
ingestion signals ─────POST /ingest/signals───────► operations read model
dashboard ◄── tRPC snapshot + authenticated SSE ───────┘
```
Direct crawls stream article drafts through the pipeline in one process. Queued execution separates discovery and article collection with BullMQ:
```text
schedule → discovery queue → worker → article queue → worker → outbox → API
```
The crawler owns collection, retries, queues, and its local durability boundary. The Basango API owns canonical article storage, signal projection, and persistent operational history. The dashboard reads the projection; it does not interpret crawler messages itself.
## Prerequisites
- Rust 1.85 or newer
- Redis for `schedule` and `worker`
- A Basango API endpoint and crawler token for article delivery and dashboard events
Redis and the API are optional for direct offline collection. Without an API endpoint, articles remain pending in SQLite until `deliver` is run later.
## Installation
```bash
git clone https://github.com/bernard-ng/basango-rs.git
cd basango-rs
cp .env.example .env
cargo build --release
```
The release binary is written to `target/release/crawler`.
## Configuration
Configuration is loaded in this order:
```text
bundled config/crawler.json < optional external JSON file < environment variables
```
The bundled JSON contains the source catalog and default HTTP, queue, runtime, and storage settings. Use `--config` or `BASANGO_CRAWLER_CONFIG_PATH` to load a different file.
Configuration is organized by capability under `src/config/`. `zod-rs` validates the raw JSON structure with strict, composable schemas and path-aware errors before Serde creates the Rust types. The same schema validates programmatically constructed configurations after environment overrides. A separate semantic validator only owns relationships that cannot be expressed as field schemas, such as distinct queue names, conditional ingestion credentials, backoff ordering, and unique source IDs.
### Ingestion API and monitoring
```bash
# Base URL of the Basango API, without an endpoint suffix
BASANGO_API_CRAWLER_ENDPOINT=https://api.basango.example
BASANGO_API_CRAWLER_TOKEN=replace-with-the-api-crawler-token
# Stable agent name displayed in ingestion operations; defaults to the hostname
BASANGO_CRAWLER_AGENT_ID=crawler-lubumbashi-01
```
With the API configured, the crawler sends:
- articles to `POST /ingest/articles`;
- lifecycle signals and heartbeats to `POST /ingest/signals`;
- update-window lookups to `POST /ingest/sources/publication-bounds`.
Article collection continues if signal reporting is temporarily unavailable. Failed article delivery remains durable in the SQLite outbox. `BASANGO_CRAWLER_NODE_ID` remains a temporary compatibility alias for the renamed agent ID variable.
### Signal protocol
The crawler emits a small discriminated protocol instead of loosely shaped event payloads:
| Signal | Meaning |
|---|---|
| `agent.heartbeat` | The worker process is reachable |
| `run.preparing` | A direct source run is resolving its inputs |
| `run.started` | Collection has started |
| `run.progress` | Absolute discovered, persisted, delivered, and failed totals |
| `run.completed` | The run completed with final totals and duration |
| `run.failed` | The run stopped with final totals, duration, and an error |
Every message has a UUID `signalId`, stable `agentId`, and `emittedAt` timestamp. Run messages also carry `runId` and `sourceId`. The API deduplicates by signal ID and projects absolute totals, so retries cannot double-count work.
### Storage and HTTP
```bash
BASANGO_CRAWLER_DATA_PATH=data
BASANGO_CRAWLER_SQLITE_PATH=data/crawler.db
BASANGO_CRAWLER_FETCH_USER_AGENT="Basango/0.1 (+https://basango.ngandu.dev)"
BASANGO_CRAWLER_FETCH_MAX_RETRIES=3
BASANGO_CRAWLER_FETCH_RESPECT_RETRY_AFTER=true
BASANGO_CRAWLER_UPDATE_DIRECTION=forward
```
### Queues
```bash
BASANGO_CRAWLER_REDIS_URL=redis://localhost:6379/0
BASANGO_CRAWLER_QUEUE_DISCOVERY=discovery
BASANGO_CRAWLER_QUEUE_ARTICLES=articles
BASANGO_CRAWLER_RETAIN_COMPLETED=3600
BASANGO_CRAWLER_RETAIN_FAILED=86400
```
See [`.env.example`](.env.example) and [`config/crawler.json`](config/crawler.json) for all supported values and source examples.
## Usage
### Direct crawling
Use direct mode for one source, backfills, debugging, and one-off collection:
```bash
cargo run -- crawl --source-id radiookapi.net
cargo run -- crawl --source-id radiookapi.net --page-range 1:5
cargo run -- crawl --source-id radiookapi.net --date-range 2025-01-01:2025-01-31
cargo run -- crawl --source-id 7sur7.cd --category politique
```
### Queued crawling
Schedule one or more sources in BullMQ:
```bash
cargo run -- schedule --source-id radiookapi.net
cargo run -- schedule --source-id radiookapi.net --source-id 7sur7.cd
```
Start workers for both queues:
```bash
cargo run -- worker
cargo run -- worker --concurrency 5
cargo run -- worker --queue discovery --queue articles --concurrency 5
```
Workers publish an agent heartbeat every 15 seconds. Stop them gracefully with `Ctrl-C`.
### Delivering the outbox
Retry pending or failed API deliveries:
```bash
cargo run -- deliver --limit 100
cargo run -- deliver --source-id radiookapi.net --limit 50
```
### External configuration
Every command accepts a configuration override:
```bash
cargo run -- --config /path/to/crawler.json crawl --source-id radiookapi.net
```
## CLI reference
| Command | Purpose |
|---|---|
| `crawl` (`sync`) | Collect one source immediately |
| `schedule` | Enqueue one or more source discovery jobs |
| `worker` | Process discovery and article queues |
| `deliver` (`push`) | Retry durable outbox deliveries |
| `version` | Print the crawler version |
Common crawl flags:
| Option | Description | Example |
|---|---|---|
| `--source-id` | Source identifier from the active configuration | `--source-id radiookapi.net` |
| `--page-range` | Inclusive page range | `--page-range 1:5` |
| `--date-range` | Inclusive UTC date window | `--date-range 2025-01-01:2025-01-31` |
| `--category` | Configured source category | `--category politique` |
## Realtime ingestion operations
Run the Basango database migration and open **Ingestion** in the Basango dashboard. The operations view shows:
- online and offline crawler agents;
- active and recent direct crawl runs;
- discovered, persisted, delivered, and failed article counts;
- failures and their latest errors;
- a live lifecycle activity feed.
The dashboard receives lightweight server-sent invalidations and reloads the durable tRPC snapshot. Periodic polling covers reconnects and multi-instance deployments. An agent is considered offline after 45 seconds without a heartbeat or lifecycle signal.
## Deployment
Example systemd units for the scheduler and worker are in [`deploy/`](deploy/). Production crawler processes are intentionally managed from this repository, not from the TypeScript monorepo's PM2 configuration.
Build and validate before deployment:
```bash
cargo fmt --check
cargo test --all-targets
cargo clippy --all-targets --all-features -- -D warnings
cargo build --release
```
## Migration from the TypeScript crawler
The TypeScript implementation has been removed from `basango/apps/crawler`. Operational ownership now lives here:
- source configuration: `config/crawler.json`;
- crawler processes and scheduling: this binary and `deploy/` units;
- ingestion and signal contracts: `@basango/domain` in the Basango monorepo;
- signal projection and operations queries: `@basango/db` and the Basango API;
- run visibility: the Basango dashboard's **Ingestion** page.
The former TypeScript event names and `/crawler/events` route are intentionally not compatibility aliases. Rust and the API now share the `agent.*` / `run.*` vocabulary above.
+38
View File
@@ -0,0 +1,38 @@
{
"lockfileVersion": 1,
"configVersion": 1,
"workspaces": {
"": {
"dependencies": {
"log": "^6.3.2",
},
},
},
"packages": {
"d": ["d@1.0.2", "", { "dependencies": { "es5-ext": "^0.10.64", "type": "^2.7.2" } }, "sha512-MOqHvMWF9/9MX6nza0KgvFH4HpMU0EF5uUDXqX/BtxtU8NfB0QzRtJ8Oe/6SuS4kbhyzVJwjd97EA4PKrzJ8bw=="],
"duration": ["duration@0.2.2", "", { "dependencies": { "d": "1", "es5-ext": "~0.10.46" } }, "sha512-06kgtea+bGreF5eKYgI/36A6pLXggY7oR4p1pq4SmdFBn1ReOL5D8RhG64VrqfTTKNucqqtBAwEj8aB88mcqrg=="],
"es5-ext": ["es5-ext@0.10.64", "", { "dependencies": { "es6-iterator": "^2.0.3", "es6-symbol": "^3.1.3", "esniff": "^2.0.1", "next-tick": "^1.1.0" } }, "sha512-p2snDhiLaXe6dahss1LddxqEm+SkuDvV8dnIQG0MWjyHpcMNfXKPE+/Cc0y+PhxJX3A4xGNeFCj5oc0BUh6deg=="],
"es6-iterator": ["es6-iterator@2.0.3", "", { "dependencies": { "d": "1", "es5-ext": "^0.10.35", "es6-symbol": "^3.1.1" } }, "sha512-zw4SRzoUkd+cl+ZoE15A9o1oQd920Bb0iOJMQkQhl3jNc03YqVjAhG7scf9C5KWRU/R13Orf588uCC6525o02g=="],
"es6-symbol": ["es6-symbol@3.1.4", "", { "dependencies": { "d": "^1.0.2", "ext": "^1.7.0" } }, "sha512-U9bFFjX8tFiATgtkJ1zg25+KviIXpgRvRHS8sau3GfhVzThRQrOeksPeT0BWW2MNZs1OEWJ1DPXOQMn0KKRkvg=="],
"esniff": ["esniff@2.0.1", "", { "dependencies": { "d": "^1.0.1", "es5-ext": "^0.10.62", "event-emitter": "^0.3.5", "type": "^2.7.2" } }, "sha512-kTUIGKQ/mDPFoJ0oVfcmyJn4iBDRptjNVIzwIFR7tqWXdVI9xfA2RMwY/gbSpJG3lkdWNEjLap/NqVHZiJsdfg=="],
"event-emitter": ["event-emitter@0.3.5", "", { "dependencies": { "d": "1", "es5-ext": "~0.10.14" } }, "sha512-D9rRn9y7kLPnJ+hMq7S/nhvoKwwvVJahBi2BPmx3bvbsEdK3W9ii8cBSGjP+72/LnM4n6fo3+dkCX5FeTQruXA=="],
"ext": ["ext@1.7.0", "", { "dependencies": { "type": "^2.7.2" } }, "sha512-6hxeJYaL110a9b5TEJSj0gojyHQAmA2ch5Os+ySCiA1QGdS697XWY1pzsrSjqA9LDEEgdB/KypIlR59RcLuHYw=="],
"log": ["log@6.3.2", "", { "dependencies": { "d": "^1.0.2", "duration": "^0.2.2", "es5-ext": "^0.10.64", "event-emitter": "^0.3.5", "sprintf-kit": "^2.0.2", "type": "^2.7.3", "uni-global": "^1.0.0" } }, "sha512-ek8NRg/OPvS9ISOJNWNAz5vZcpYacWNFDWNJjj5OXsc6YuKacfey6wF04cXz/tOJIVrZ2nGSkHpAY5qKtF6ISg=="],
"next-tick": ["next-tick@1.1.0", "", {}, "sha512-CXdUiJembsNjuToQvxayPZF9Vqht7hewsvy2sOWafLvi2awflj9mOC6bHIg50orX8IJvWKY9wYQ/zB2kogPslQ=="],
"sprintf-kit": ["sprintf-kit@2.0.2", "", { "dependencies": { "es5-ext": "^0.10.64" } }, "sha512-lnapdj6W4LflHZGKvl9eVkz5YF0xaTrqpRWVA4cNVOTedwqifIP8ooGImldzT/4IAN5KXFQAyXTdLidYVQdyag=="],
"type": ["type@2.7.3", "", {}, "sha512-8j+1QmAbPvLZow5Qpi6NCaN8FB60p/6x8/vfNqOk/hC+HuvFZhL4+WfekuhQLiqFZXOgQdrs3B+XxEmCc6b3FQ=="],
"uni-global": ["uni-global@1.0.0", "", { "dependencies": { "type": "^2.5.0" } }, "sha512-WWM3HP+siTxzIWPNUg7hZ4XO8clKi6NoCAJJWnuRL+BAqyFXF8gC03WNyTefGoUXYc47uYgXxpKLIEvo65PEHw=="],
}
}
+235
View File
@@ -0,0 +1,235 @@
{
"queue": {
"prefix": "basango:crawler",
"queues": {
"articles": "articles",
"discovery": "discovery"
},
"redis_url": "redis://localhost:6379/0",
"retention": {
"completed": 3600,
"failed": 86400
}
},
"http": {
"backoff": {
"initial": 1,
"max": 30,
"multiplier": 2
},
"max_retries": 3,
"rotate": true,
"timeout": 20,
"user_agent": "Basango/0.1 (+https://basango.ngandu.dev)"
},
"runtime": {
"direction": "forward",
"worker_concurrency": 5
},
"sources": [
{
"kind": "html",
"id": "radiookapi.net",
"url": "https://www.radiookapi.net",
"fetch_details": true,
"pagination_template": "actualite",
"selectors": {
"body": ".field-name-body",
"categories": ".views-field-field-cat-gorie a",
"date": "head > meta[property=\"article:published_time\"]",
"link": ".views-field-title a",
"list": ".view-content > .views-row.content-row",
"title": "h1.page-header",
"pagination": "ul.pagination > li.pager-last > a"
}
},
{
"kind": "html",
"id": "7sur7.cd",
"url": "https://7sur7.cd",
"fetch_details": true,
"pagination_template": "index.php/category/{category}",
"selectors": {
"body": "div[property=\"schema:text\"].field.field--name-body",
"date": "head > meta[property=\"article:published_time\"]",
"link": ".views-field-title a",
"list": ".view-content > .row.views-row",
"title": ".views-field-title a",
"pagination": "ul.pagination > li.pager__item.pager__item--last > a"
}
},
{
"kind": "html",
"id": "mediacongo.net",
"url": "https://www.mediacongo.net",
"date_format": "dd.MM.yyyy",
"fetch_details": true,
"pagination_template": "articles.html",
"selectors": {
"body": ".article_ttext",
"categories": "a.color_link",
"date": ".article_other_about",
"link": "a:first-child",
"list": ".for_aitems > .article_other_item",
"title": "h1",
"pagination": "div.pagination > div > a:last-child"
}
},
{
"kind": "html",
"id": "actualite.cd",
"url": "https://actualite.cd",
"fetch_details": true,
"pagination_template": "actualite",
"selectors": {
"body": ".views-field.views-field-body .field-content",
"categories": "#actu-cat",
"date": "head > meta[property=\"article:published_time\"]",
"link": "#actu-titre a",
"list": "#views-bootstrap-taxonomy-term-page-2 > div > div",
"title": "h1.page-title"
}
},
{
"kind": "wordpress",
"id": "beto.cd",
"url": "https://beto.cd",
"rate_limit": true
},
{
"kind": "wordpress",
"id": "newscd.net",
"url": "https://newscd.net"
},
{
"kind": "wordpress",
"id": "africanewsrdc.net",
"url": "https://www.africanewsrdc.net"
},
{
"kind": "wordpress",
"id": "angazainstitute.ac.cd",
"url": "https://angazainstitute.ac.cd"
},
{
"kind": "wordpress",
"id": "b-onetv.cd",
"url": "https://b-onetv.cd"
},
{
"kind": "wordpress",
"id": "bukavufm.com",
"url": "https://bukavufm.com"
},
{
"kind": "wordpress",
"id": "changement7.net",
"url": "https://changement7.net"
},
{
"kind": "wordpress",
"id": "congoactu.net",
"url": "https://congoactu.net"
},
{
"kind": "wordpress",
"id": "congoindependant.com",
"url": "https://www.congoindependant.com"
},
{
"kind": "wordpress",
"id": "congoquotidien.com",
"url": "https://www.congoquotidien.com"
},
{
"kind": "wordpress",
"id": "cumulard.cd",
"url": "https://www.cumulard.cd"
},
{
"kind": "wordpress",
"id": "environews-rdc.net",
"url": "https://environews-rdc.net"
},
{
"kind": "wordpress",
"id": "freemediardc.info",
"url": "https://www.freemediardc.info"
},
{
"kind": "wordpress",
"id": "geopolismagazine.org",
"url": "https://geopolismagazine.org"
},
{
"kind": "wordpress",
"id": "habarirdc.net",
"url": "https://habarirdc.net"
},
{
"kind": "wordpress",
"id": "infordc.com",
"url": "https://infordc.com"
},
{
"kind": "wordpress",
"id": "kilalopress.net",
"url": "https://kilalopress.net"
},
{
"kind": "wordpress",
"id": "laprosperiteonline.net",
"url": "https://laprosperiteonline.net"
},
{
"kind": "wordpress",
"id": "laprunellerdc.cd",
"url": "https://laprunellerdc.cd"
},
{
"kind": "wordpress",
"id": "lesmedias.net",
"url": "https://lesmedias.net"
},
{
"kind": "wordpress",
"id": "lesvolcansnews.net",
"url": "https://lesvolcansnews.net"
},
{
"kind": "wordpress",
"id": "netic-news.net",
"url": "https://www.netic-news.net"
},
{
"kind": "wordpress",
"id": "objectif-infos.cd",
"url": "https://objectif-infos.cd"
},
{
"kind": "wordpress",
"id": "scooprdc.net",
"url": "https://scooprdc.net"
},
{
"kind": "wordpress",
"id": "journaldekinshasa.com",
"url": "https://www.journaldekinshasa.com"
},
{
"kind": "wordpress",
"id": "lepotentiel.cd",
"url": "https://lepotentiel.cd"
},
{
"kind": "wordpress",
"id": "acturdc.com",
"url": "https://acturdc.com"
},
{
"kind": "wordpress",
"id": "matininfos.net",
"url": "https://matininfos.net"
}
]
}
+17
View File
@@ -0,0 +1,17 @@
# Binary deployment
Build an optimized binary:
```bash
cargo build --release
```
Install `target/release/crawler` and `.env` under `/opt/crawler`. The default
source configuration is already embedded in the binary. Store the SQLite
outbox under `/var/lib/crawler` by setting:
```text
BASANGO_CRAWLER_SQLITE_PATH=/var/lib/crawler/crawler.db
```
Then install the three systemd unit files in this directory, reload systemd, and enable the worker and timer. The scheduler reads `BASANGO_CRAWLER_SOURCE_IDS`, allowing each machine to own a different source shard.
+14
View File
@@ -0,0 +1,14 @@
# One-shot scheduler. The timer below invokes this unit periodically.
[Unit]
Description=Schedule Basango Rust crawler jobs for this node
After=network-online.target
Wants=network-online.target
[Service]
Type=oneshot
WorkingDirectory=/opt/crawler
EnvironmentFile=/opt/crawler/.env
ExecStart=/opt/crawler/crawler schedule
User=basango
Group=basango
StateDirectory=crawler
+13
View File
@@ -0,0 +1,13 @@
# Timers provide persistent scheduling without embedding a cron loop in Rust.
[Unit]
Description=Run the Basango Rust crawler scheduler periodically
[Timer]
OnBootSec=2min
OnUnitActiveSec=30min
RandomizedDelaySec=5min
Persistent=true
Unit=crawler-schedule.service
[Install]
WantedBy=timers.target
+19
View File
@@ -0,0 +1,19 @@
# Long-running worker. Systemd restarts it after infrastructure failures.
[Unit]
Description=Basango Rust crawler worker
After=network-online.target
Wants=network-online.target
[Service]
Type=simple
WorkingDirectory=/opt/crawler
EnvironmentFile=/opt/crawler/.env
ExecStart=/opt/crawler/crawler worker --queue discovery --queue articles
Restart=always
RestartSec=10
User=basango
Group=basango
StateDirectory=crawler
[Install]
WantedBy=multi-user.target
+64
View File
@@ -0,0 +1,64 @@
//! Article ingestion pipeline.
//!
//! A crawler first produces an [`ArticleDraft`].
//! This module then normalizes it, durably stores it, and optionally forwards
//! it. Persisting before network delivery is the *outbox pattern*: a process
//! crash cannot silently lose an article that was already collected.
mod forwarder;
mod normalize;
mod outbox;
pub(crate) use forwarder::endpoint_url;
pub use forwarder::{ArticleIngestionClient, DeliveryResult};
pub use normalize::normalize;
pub use outbox::{DeliveryStatus, Outbox, OutboxEntry};
use crate::{
domain::{Article, ArticleDraft},
error::Result,
};
/// What happened when a draft entered the durable ingestion pipeline.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum IngestStatus {
/// It was saved but no backend is configured, so `deliver` can send it later.
Persisted,
/// It had already reached the backend in a previous attempt.
AlreadyForwarded,
/// It was saved and forwarded during this call.
Forwarded,
/// Forwarding failed; the outbox retains error and retry information.
DeliveryFailed,
}
/// Normalize, persist, and—when configured—forward one article.
pub async fn ingest(
draft: ArticleDraft,
outbox: &Outbox,
ingestion: Option<&ArticleIngestionClient>,
) -> Result<(Article, IngestStatus)> {
let article = normalize(draft)?;
let status = outbox.save(&article)?;
if status == DeliveryStatus::Forwarded {
return Ok((article, IngestStatus::AlreadyForwarded));
}
let Some(ingestion) = ingestion else {
return Ok((article, IngestStatus::Persisted));
};
// `Outbox` releases its synchronous lock before this network await.
match ingestion.deliver(&article).await {
DeliveryResult::Delivered { .. } => {
outbox.mark_forwarded(&article.hash)?;
Ok((article, IngestStatus::Forwarded))
}
DeliveryResult::Failed {
retryable, message, ..
} => {
outbox.mark_failed(&article.hash, &message, retryable)?;
Ok((article, IngestStatus::DeliveryFailed))
}
}
}
+111
View File
@@ -0,0 +1,111 @@
//! Delivery of persisted articles to the Basango ingestion API.
use url::Url;
use crate::{config::IngestionApiConfig, domain::Article, error::Result, http::HttpClient};
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum DeliveryResult {
Delivered {
status: u16,
},
Failed {
retryable: bool,
status: Option<u16>,
message: String,
},
}
#[derive(Clone)]
pub struct ArticleIngestionClient {
client: HttpClient,
endpoint: Url,
token: String,
}
impl ArticleIngestionClient {
pub fn new(config: &IngestionApiConfig, client: HttpClient) -> Result<Option<Self>> {
let Some(endpoint) = config.endpoint.clone() else {
return Ok(None);
};
Ok(Some(Self {
client,
endpoint,
token: config.token.clone(),
}))
}
pub async fn deliver(&self, article: &Article) -> DeliveryResult {
let endpoint = match endpoint_url(&self.endpoint, "ingest/articles") {
Ok(endpoint) => endpoint,
Err(error) => {
return DeliveryResult::Failed {
retryable: false,
status: None,
message: error.to_string(),
};
}
};
let headers = [
("Authorization", self.token.as_str()),
("Idempotency-Key", article.hash.as_str()),
];
match self.client.post_json(&endpoint, &headers, article).await {
Ok(response) if response.is_success() => DeliveryResult::Delivered {
status: response.status.as_u16(),
},
Ok(response) => {
let status = response.status.as_u16();
DeliveryResult::Failed {
retryable: is_retryable_status(status),
status: Some(status),
message: format!(
"forwarding failed with HTTP {status}: {}",
response.body_lossy()
),
}
}
Err(error) => DeliveryResult::Failed {
retryable: true,
status: None,
message: error.to_string(),
},
}
}
}
/// Append a path segment without `Url::join`'s “replace the final segment”
/// behavior. API base URLs often contain a path such as `/crawler`.
pub(crate) fn endpoint_url(base: &Url, segment: &str) -> Result<Url> {
let mut result = base.clone();
let mut path = result.path().trim_end_matches('/').to_owned();
path.push('/');
path.push_str(segment.trim_start_matches('/'));
result.set_path(&path);
Ok(result)
}
fn is_retryable_status(status: u16) -> bool {
matches!(status, 408 | 425 | 429) || status >= 500
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn appends_endpoint_to_an_existing_base_path() {
let base = Url::parse("https://api.example.com/crawler").unwrap();
assert_eq!(
endpoint_url(&base, "ingest/articles").unwrap().as_str(),
"https://api.example.com/crawler/ingest/articles"
);
}
#[test]
fn retryable_statuses_match_transport_semantics() {
assert!(is_retryable_status(503));
assert!(!is_retryable_status(422));
}
}
+94
View File
@@ -0,0 +1,94 @@
//! Validation and canonicalization at the article boundary.
use std::collections::HashSet;
use crate::{
domain::{Article, ArticleDraft},
error::{CrawlError, Result},
};
/// Convert a permissive source draft into the canonical stored representation.
pub fn normalize(draft: ArticleDraft) -> Result<Article> {
let title = sanitize(&draft.title);
let body = sanitize(&draft.body);
if title.is_empty() {
return Err(CrawlError::InvalidArticle("title cannot be empty".into()));
}
if body.is_empty() {
return Err(CrawlError::InvalidArticle("body cannot be empty".into()));
}
// The URL hash is an identity value, not a security boundary. It lets
// SQLite enforce idempotency when a URL is crawled more than once.
let hash = format!("{:x}", md5::compute(draft.link.as_str()));
let mut seen = HashSet::new();
let categories = draft
.categories
.iter()
.map(|category| sanitize(category))
.filter(|category| !category.is_empty())
.filter(|category| seen.insert(category.to_lowercase()))
.collect();
Ok(Article {
hash,
title,
body,
link: draft.link,
source_id: draft.source_id,
categories,
metadata: draft.metadata.filter(|metadata| !metadata.is_empty()),
published_at: draft.published_at,
})
}
fn sanitize(text: &str) -> String {
// These invisible Unicode characters commonly arrive in copied news text.
// Normalizing them once protects all output adapters.
let normalized = text
.replace(['\u{00a0}', '\u{202f}'], " ")
.replace(['\u{200b}', '\u{200c}', '\u{200d}', '\u{feff}'], "")
.replace("\r\n", "\n");
let mut result = String::with_capacity(normalized.len());
let mut previous_newline = false;
for character in normalized.chars() {
if character == '\n' {
if !previous_newline {
result.push(character);
}
previous_newline = true;
} else {
result.push(character);
previous_newline = false;
}
}
result.trim().to_owned()
}
#[cfg(test)]
mod tests {
use chrono::Utc;
use url::Url;
use super::*;
#[test]
fn normalizes_text_and_deduplicates_categories() {
let article = normalize(ArticleDraft {
title: " A\u{00a0}title ".into(),
body: "body\n\n\nsecond".into(),
link: Url::parse("https://example.com/article").unwrap(),
source_id: crate::domain::SourceId::new("example").unwrap(),
categories: vec!["News".into(), "news".into()],
metadata: None,
published_at: Utc::now(),
})
.unwrap();
assert_eq!(article.title, "A title");
assert_eq!(article.body, "body\nsecond");
assert_eq!(article.categories, vec!["News"]);
assert_eq!(article.hash.len(), 32);
}
}
+420
View File
@@ -0,0 +1,420 @@
//! Durable SQLite article outbox.
//!
//! SQLite is synchronous, so these methods are intentionally short and no lock
//! is held across `.await`. Clones share one connection inside a process; WAL
//! mode still allows other crawler processes to coexist safely.
use std::{
path::Path,
str::FromStr,
sync::{Arc, Mutex, MutexGuard},
};
use chrono::{DateTime, Duration, Utc};
use rusqlite::{
Connection, OptionalExtension, TransactionBehavior, named_params, params, types::Type,
};
use crate::{
domain::Article,
error::{CrawlError, Result},
};
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum DeliveryStatus {
Pending,
Forwarded,
Failed,
}
impl FromStr for DeliveryStatus {
type Err = CrawlError;
fn from_str(value: &str) -> Result<Self> {
match value {
"pending" => Ok(Self::Pending),
"forwarded" => Ok(Self::Forwarded),
"failed" => Ok(Self::Failed),
other => Err(CrawlError::Configuration(format!(
"unknown outbox status '{other}'"
))),
}
}
}
#[derive(Debug, Clone)]
pub struct OutboxEntry {
pub article: Article,
pub status: DeliveryStatus,
pub attempts: u32,
pub retryable: bool,
pub last_error: Option<String>,
pub created_at: DateTime<Utc>,
pub updated_at: DateTime<Utc>,
pub forwarded_at: Option<DateTime<Utc>>,
pub claimed_at: Option<DateTime<Utc>>,
pub claimed_by: Option<String>,
}
#[derive(Clone)]
pub struct Outbox {
connection: Arc<Mutex<Connection>>,
}
impl Outbox {
pub fn open(path: &Path, create: bool) -> Result<Self> {
if !create && !path.exists() {
return Err(CrawlError::Configuration(format!(
"SQLite outbox does not exist: {}",
path.display()
)));
}
if create
&& let Some(parent) = path.parent()
&& !parent.as_os_str().is_empty()
{
std::fs::create_dir_all(parent)?;
}
let connection = Connection::open(path)?;
let outbox = Self {
connection: Arc::new(Mutex::new(connection)),
};
outbox.migrate()?;
Ok(outbox)
}
pub fn exists(path: &Path) -> bool {
path.exists()
}
/// Upsert by hash while preserving an already-forwarded state. This is the
/// idempotency boundary: crawling the same URL twice does not redeliver it.
pub fn save(&self, article: &Article) -> Result<DeliveryStatus> {
let timestamp = Utc::now().to_rfc3339();
let categories = serde_json::to_string(&article.categories)?;
let metadata = article
.metadata
.as_ref()
.map(serde_json::to_string)
.transpose()?;
let payload = serde_json::to_string(article)?;
let connection = self.connection()?;
connection.execute(
r#"
INSERT INTO articles (
hash, source_id, link, title, body, categories, metadata,
published_at, payload, status, attempts, retryable,
last_error, created_at, updated_at, forwarded_at
)
VALUES (
:hash, :source_id, :link, :title, :body, :categories, :metadata,
:published_at, :payload, 'pending', 0, 1, NULL, :now, :now, NULL
)
ON CONFLICT(hash) DO UPDATE SET
source_id = excluded.source_id,
link = excluded.link,
title = excluded.title,
body = excluded.body,
categories = excluded.categories,
metadata = excluded.metadata,
published_at = excluded.published_at,
payload = excluded.payload,
status = CASE WHEN articles.status = 'forwarded' THEN 'forwarded' ELSE 'pending' END,
last_error = CASE WHEN articles.status = 'forwarded' THEN articles.last_error ELSE NULL END,
retryable = CASE WHEN articles.status = 'forwarded' THEN articles.retryable ELSE 1 END,
updated_at = excluded.updated_at,
forwarded_at = CASE WHEN articles.status = 'forwarded' THEN articles.forwarded_at ELSE NULL END,
claimed_at = NULL,
claimed_by = NULL
"#,
named_params! {
":hash": article.hash,
":source_id": article.source_id.as_str(),
":link": article.link.as_str(),
":title": article.title,
":body": article.body,
":categories": categories,
":metadata": metadata,
":published_at": article.published_at.to_rfc3339(),
":payload": payload,
":now": timestamp,
},
)?;
let status: String = connection.query_row(
"SELECT status FROM articles WHERE hash = ?1",
[&article.hash],
|row| row.get(0),
)?;
status.parse()
}
pub fn list_pending(&self, source_id: Option<&str>, limit: usize) -> Result<Vec<OutboxEntry>> {
let sql = if source_id.is_some() {
r#"SELECT payload, status, attempts, retryable, last_error, created_at,
updated_at, forwarded_at, claimed_at, claimed_by
FROM articles
WHERE status IN ('pending', 'failed') AND retryable = 1 AND source_id = ?1
ORDER BY created_at ASC LIMIT ?2"#
} else {
r#"SELECT payload, status, attempts, retryable, last_error, created_at,
updated_at, forwarded_at, claimed_at, claimed_by
FROM articles
WHERE status IN ('pending', 'failed') AND retryable = 1
ORDER BY created_at ASC LIMIT ?1"#
};
let connection = self.connection()?;
let mut statement = connection.prepare(sql)?;
let rows = if let Some(source_id) = source_id {
statement.query_map(params![source_id, limit as i64], row_to_article)?
} else {
statement.query_map([limit as i64], row_to_article)?
};
rows.collect::<std::result::Result<Vec<_>, _>>()
.map_err(Into::into)
}
/// Atomically reserve a batch for one pusher.
///
/// `IMMEDIATE` obtains SQLite's write reservation before selecting rows.
/// Without that transaction boundary, two processes could select and send
/// the same pending articles concurrently.
pub fn claim(
&self,
claimed_by: &str,
source_id: Option<&str>,
limit: usize,
claim_ttl: Duration,
) -> Result<Vec<OutboxEntry>> {
let now = Utc::now();
let expires_before = now - claim_ttl;
let mut connection = self.connection()?;
let transaction = connection.transaction_with_behavior(TransactionBehavior::Immediate)?;
let select = if source_id.is_some() {
r#"SELECT hash FROM articles
WHERE status IN ('pending', 'failed') AND retryable = 1
AND source_id = ?1 AND (claimed_at IS NULL OR claimed_at < ?2)
ORDER BY created_at ASC LIMIT ?3"#
} else {
r#"SELECT hash FROM articles
WHERE status IN ('pending', 'failed') AND retryable = 1
AND (claimed_at IS NULL OR claimed_at < ?1)
ORDER BY created_at ASC LIMIT ?2"#
};
let hashes = {
let mut statement = transaction.prepare(select)?;
if let Some(source_id) = source_id {
statement
.query_map(
params![source_id, expires_before.to_rfc3339(), limit as i64],
|row| row.get::<_, String>(0),
)?
.collect::<std::result::Result<Vec<_>, _>>()?
} else {
statement
.query_map(params![expires_before.to_rfc3339(), limit as i64], |row| {
row.get::<_, String>(0)
})?
.collect::<std::result::Result<Vec<_>, _>>()?
}
};
for hash in &hashes {
transaction.execute(
"UPDATE articles SET claimed_at = ?1, claimed_by = ?2, updated_at = ?1 WHERE hash = ?3",
params![now.to_rfc3339(), claimed_by, hash],
)?;
}
transaction.commit()?;
drop(connection);
hashes
.iter()
.map(|hash| {
self.get(hash)?
.ok_or_else(|| CrawlError::Queue("claimed article disappeared".into()))
})
.collect()
}
pub fn mark_forwarded(&self, hash: &str) -> Result<()> {
let now = Utc::now().to_rfc3339();
self.connection()?.execute(
r#"UPDATE articles SET status = 'forwarded', last_error = NULL,
retryable = 0, updated_at = ?1, forwarded_at = ?1,
claimed_at = NULL, claimed_by = NULL WHERE hash = ?2"#,
params![now, hash],
)?;
Ok(())
}
pub fn mark_failed(&self, hash: &str, error: &str, retryable: bool) -> Result<()> {
self.connection()?.execute(
r#"UPDATE articles SET status = 'failed', attempts = attempts + 1,
retryable = ?1, last_error = ?2, updated_at = ?3,
claimed_at = NULL, claimed_by = NULL WHERE hash = ?4"#,
params![retryable, error, Utc::now().to_rfc3339(), hash],
)?;
Ok(())
}
pub fn get(&self, hash: &str) -> Result<Option<OutboxEntry>> {
self.connection()?
.query_row(
r#"SELECT payload, status, attempts, retryable, last_error, created_at,
updated_at, forwarded_at, claimed_at, claimed_by
FROM articles WHERE hash = ?1"#,
[hash],
row_to_article,
)
.optional()
.map_err(Into::into)
}
fn migrate(&self) -> Result<()> {
self.connection()?.execute_batch(
r#"
PRAGMA journal_mode = WAL;
PRAGMA synchronous = NORMAL;
PRAGMA busy_timeout = 5000;
PRAGMA foreign_keys = ON;
CREATE TABLE IF NOT EXISTS articles (
hash TEXT PRIMARY KEY,
source_id TEXT NOT NULL,
link TEXT NOT NULL UNIQUE,
title TEXT NOT NULL,
body TEXT NOT NULL,
categories TEXT NOT NULL DEFAULT '[]',
metadata TEXT,
published_at TEXT NOT NULL,
payload TEXT NOT NULL,
status TEXT NOT NULL DEFAULT 'pending'
CHECK (status IN ('pending', 'forwarded', 'failed')),
attempts INTEGER NOT NULL DEFAULT 0,
retryable INTEGER NOT NULL DEFAULT 1,
last_error TEXT,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL,
forwarded_at TEXT,
claimed_at TEXT,
claimed_by TEXT
);
CREATE INDEX IF NOT EXISTS articles_status_created_at_idx
ON articles(status, created_at);
CREATE INDEX IF NOT EXISTS articles_source_status_idx
ON articles(source_id, status);
CREATE INDEX IF NOT EXISTS articles_claimed_at_created_at_idx
ON articles(claimed_at, created_at);
"#,
)?;
Ok(())
}
fn connection(&self) -> Result<MutexGuard<'_, Connection>> {
self.connection
.lock()
.map_err(|_| CrawlError::OutboxLockPoisoned)
}
}
fn row_to_article(row: &rusqlite::Row<'_>) -> rusqlite::Result<OutboxEntry> {
let payload: String = row.get(0)?;
let status: String = row.get(1)?;
Ok(OutboxEntry {
article: serde_json::from_str(&payload).map_err(|error| sql_conversion(0, error))?,
status: status.parse().map_err(|error| sql_conversion(1, error))?,
attempts: row.get(2)?,
retryable: row.get(3)?,
last_error: row.get(4)?,
created_at: parse_sql_date(row, 5)?,
updated_at: parse_sql_date(row, 6)?,
forwarded_at: parse_optional_sql_date(row, 7)?,
claimed_at: parse_optional_sql_date(row, 8)?,
claimed_by: row.get(9)?,
})
}
fn parse_sql_date(row: &rusqlite::Row<'_>, index: usize) -> rusqlite::Result<DateTime<Utc>> {
let value: String = row.get(index)?;
DateTime::parse_from_rfc3339(&value)
.map(|date| date.with_timezone(&Utc))
.map_err(|error| sql_conversion(index, error))
}
fn parse_optional_sql_date(
row: &rusqlite::Row<'_>,
index: usize,
) -> rusqlite::Result<Option<DateTime<Utc>>> {
let value: Option<String> = row.get(index)?;
value
.map(|value| {
DateTime::parse_from_rfc3339(&value)
.map(|date| date.with_timezone(&Utc))
.map_err(|error| sql_conversion(index, error))
})
.transpose()
}
fn sql_conversion(
index: usize,
error: impl std::error::Error + Send + Sync + 'static,
) -> rusqlite::Error {
rusqlite::Error::FromSqlConversionFailure(index, Type::Text, Box::new(error))
}
#[cfg(test)]
mod tests {
use chrono::Utc;
use tempfile::tempdir;
use url::Url;
use super::*;
fn article() -> Article {
Article {
hash: "hash-1".into(),
title: "Title".into(),
body: "Body".into(),
link: Url::parse("https://example.com/one").unwrap(),
source_id: crate::domain::SourceId::new("example").unwrap(),
categories: vec!["news".into()],
metadata: None,
published_at: Utc::now(),
}
}
#[test]
fn forwarded_rows_stay_forwarded_when_saved_again() {
let directory = tempdir().unwrap();
let path = directory.path().join("outbox.db");
let outbox = Outbox::open(&path, true).unwrap();
let article = article();
assert_eq!(outbox.save(&article).unwrap(), DeliveryStatus::Pending);
outbox.mark_forwarded(&article.hash).unwrap();
assert_eq!(outbox.save(&article).unwrap(), DeliveryStatus::Forwarded);
}
#[test]
fn claim_reserves_pending_rows() {
let directory = tempdir().unwrap();
let path = directory.path().join("outbox.db");
let outbox = Outbox::open(&path, true).unwrap();
outbox.save(&article()).unwrap();
let claimed = outbox
.claim("worker-1", None, 10, Duration::minutes(15))
.unwrap();
assert_eq!(claimed.len(), 1);
assert_eq!(claimed[0].claimed_by.as_deref(), Some("worker-1"));
let second = outbox
.claim("worker-2", None, 10, Duration::minutes(15))
.unwrap();
assert!(second.is_empty());
}
}
+234
View File
@@ -0,0 +1,234 @@
//! Command-line interface and process-level wiring.
//!
//! Clap parses untrusted strings at the outermost boundary. Commands then use
//! typed ranges and options, so deeper modules do not repeatedly validate the
//! same input.
use std::{env, path::PathBuf};
use anyhow::{Context, bail};
use clap::{Args, Parser, Subcommand};
use tracing_subscriber::EnvFilter;
use crate::{
Crawler,
domain::{CrawlRequest, DateRange, PageRange, SourceId},
};
#[derive(Debug, Parser)]
#[command(
name = "crawler",
version,
about = "Collect Congolese news from HTML and WordPress sources"
)]
struct Cli {
/// Override the bundled JSON configuration file.
#[arg(long, global = true, value_name = "PATH")]
config: Option<PathBuf>,
#[command(subcommand)]
command: Command,
}
#[derive(Debug, Subcommand)]
enum Command {
/// Crawl one source immediately in this process.
#[command(alias = "sync")]
Crawl(CrawlArgs),
/// Place one or more source discovery jobs in BullMQ.
Schedule(ScheduleArgs),
/// Process BullMQ discovery and article jobs until interrupted.
Worker(WorkerArgs),
/// Deliver pending or failed articles from the SQLite outbox.
#[command(alias = "push")]
Deliver(DeliverArgs),
/// Print version information (also available as --version).
Version,
}
#[derive(Debug, Args)]
struct CrawlArgs {
/// Source identifier from the active configuration.
#[arg(long)]
source_id: SourceId,
/// Inclusive page range in start:end form, for example 1:5.
#[arg(long, value_parser = parse_page_range)]
page_range: Option<PageRange>,
/// Inclusive UTC date range, for example 2025-01-01:2025-01-31.
#[arg(long, value_parser = parse_date_range)]
date_range: Option<DateRange>,
/// Optional configured category slug.
#[arg(long)]
category: Option<String>,
}
#[derive(Debug, Args)]
struct ScheduleArgs {
/// Repeat the flag or pass comma-separated IDs.
#[arg(long = "source-id", value_delimiter = ',')]
source_ids: Vec<SourceId>,
/// Inclusive page range in start:end form, for example 1:5.
#[arg(long, value_parser = parse_page_range)]
page_range: Option<PageRange>,
/// Inclusive UTC date range, for example 2025-01-01:2025-01-31.
#[arg(long, value_parser = parse_date_range)]
date_range: Option<DateRange>,
/// Optional configured category slug.
#[arg(long)]
category: Option<String>,
}
#[derive(Debug, Args)]
struct WorkerArgs {
/// Queue suffix to process; repeat to select both explicitly.
#[arg(long, short = 'q')]
queue: Vec<String>,
/// Maximum number of jobs processed concurrently.
#[arg(long)]
concurrency: Option<usize>,
}
#[derive(Debug, Args)]
struct DeliverArgs {
/// Only claim articles collected from this source.
#[arg(long)]
source_id: Option<SourceId>,
/// Maximum number of outbox rows to claim.
#[arg(long, default_value_t = 100, value_parser = parse_positive_usize)]
limit: usize,
}
pub async fn run() -> anyhow::Result<()> {
initialize_logging();
let cli = Cli::parse();
if matches!(cli.command, Command::Version) {
println!("crawler {}", env!("CARGO_PKG_VERSION"));
return Ok(());
}
let crawler = match cli.config {
Some(path) => Crawler::from_config_file(path),
None => Crawler::from_environment(),
}
.context("could not initialize crawler")?;
match cli.command {
Command::Crawl(arguments) => {
let report = crawler.crawl(arguments.into()).await?;
tracing::info!(?report, "crawl completed");
}
Command::Schedule(arguments) => {
schedule(&crawler, arguments).await?;
}
Command::Worker(arguments) => {
let concurrency = arguments
.concurrency
.unwrap_or(crawler.config().runtime.worker_concurrency);
crawler.work(arguments.queue, concurrency).await?;
}
Command::Deliver(arguments) => {
let report = crawler
.deliver_pending(arguments.source_id.as_ref(), arguments.limit)
.await?;
tracing::info!(?report, "outbox delivery completed");
if report.failed > 0 {
bail!("failed to deliver {} article(s)", report.failed);
}
}
Command::Version => unreachable!("handled before configuration loading"),
}
Ok(())
}
async fn schedule(crawler: &Crawler, arguments: ScheduleArgs) -> anyhow::Result<()> {
let source_ids = if arguments.source_ids.is_empty() {
env::var("BASANGO_CRAWLER_SOURCE_IDS")
.unwrap_or_default()
.split(',')
.map(str::trim)
.filter(|value| !value.is_empty())
.map(str::parse)
.collect::<Result<Vec<SourceId>, _>>()?
} else {
arguments.source_ids
};
if source_ids.is_empty() {
bail!("pass --source-id or set BASANGO_CRAWLER_SOURCE_IDS");
}
for source_id in source_ids {
let id = crawler
.schedule(CrawlRequest {
source_id: source_id.clone(),
page_range: arguments.page_range,
date_range: arguments.date_range,
category: arguments.category.clone(),
})
.await?;
tracing::info!(job_id = id, %source_id, "scheduled source discovery");
}
Ok(())
}
impl From<CrawlArgs> for CrawlRequest {
fn from(value: CrawlArgs) -> Self {
Self {
source_id: value.source_id,
page_range: value.page_range,
date_range: value.date_range,
category: value.category,
}
}
}
fn parse_page_range(value: &str) -> Result<PageRange, String> {
PageRange::parse(value).map_err(|error| error.to_string())
}
fn parse_date_range(value: &str) -> Result<DateRange, String> {
DateRange::parse(value).map_err(|error| error.to_string())
}
fn parse_positive_usize(value: &str) -> Result<usize, String> {
let parsed = value
.parse::<usize>()
.map_err(|_| format!("'{value}' is not a positive integer"))?;
if parsed == 0 {
return Err("value must be at least 1".into());
}
Ok(parsed)
}
fn initialize_logging() {
let filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info"));
// Tests or embedding applications may already have a subscriber. `try_init`
// avoids panicking when global logging was initialized elsewhere.
let _ = tracing_subscriber::fmt().with_env_filter(filter).try_init();
}
#[cfg(test)]
mod tests {
use clap::Parser;
use super::*;
#[test]
fn clap_parses_typed_crawl_ranges() {
let cli = Cli::try_parse_from([
"crawler",
"crawl",
"--source-id",
"example",
"--page-range",
"1:3",
"--date-range",
"2025-01-01:2025-01-31",
])
.unwrap();
let Command::Crawl(arguments) = cli.command else {
panic!("expected crawl command")
};
assert_eq!(arguments.page_range.unwrap(), PageRange::new(1, 3).unwrap());
}
}
+106
View File
@@ -0,0 +1,106 @@
use std::{env, str::FromStr};
use url::Url;
use crate::{
domain::UpdateDirection,
error::{CrawlError, Result},
};
use super::CrawlerConfig;
pub(super) fn apply(config: &mut CrawlerConfig) -> Result<()> {
if let Some(raw) = value("BASANGO_API_CRAWLER_ENDPOINT") {
config.ingestion.endpoint = Some(Url::parse(&raw).map_err(|error| {
CrawlError::Configuration(format!("invalid ingestion API endpoint: {error}"))
})?);
}
set_string("BASANGO_API_CRAWLER_TOKEN", &mut config.ingestion.token);
set_string("BASANGO_CRAWLER_REDIS_URL", &mut config.queue.redis_url);
set_string(
"BASANGO_CRAWLER_QUEUE_DISCOVERY",
&mut config.queue.queues.discovery,
);
set_string(
"BASANGO_CRAWLER_QUEUE_ARTICLES",
&mut config.queue.queues.articles,
);
set_parsed(
"BASANGO_CRAWLER_RETAIN_COMPLETED",
&mut config.queue.retention.completed,
)?;
set_parsed(
"BASANGO_CRAWLER_RETAIN_FAILED",
&mut config.queue.retention.failed,
)?;
set_string(
"BASANGO_CRAWLER_FETCH_USER_AGENT",
&mut config.http.user_agent,
);
set_parsed(
"BASANGO_CRAWLER_FETCH_MAX_RETRIES",
&mut config.http.max_retries,
)?;
if let Some(raw) = value("BASANGO_CRAWLER_FETCH_RESPECT_RETRY_AFTER") {
config.http.respect_retry_after =
parse_bool("BASANGO_CRAWLER_FETCH_RESPECT_RETRY_AFTER", &raw)?;
}
if let Some(raw) = value("BASANGO_CRAWLER_UPDATE_DIRECTION") {
config.runtime.direction = match raw.as_str() {
"forward" => UpdateDirection::Forward,
"backward" => UpdateDirection::Backward,
_ => {
return Err(CrawlError::Configuration(format!(
"BASANGO_CRAWLER_UPDATE_DIRECTION must be 'forward' or 'backward', got '{raw}'"
)));
}
};
}
if let Some(raw) = value("BASANGO_CRAWLER_DATA_PATH") {
config.paths.data = raw.into();
}
if let Some(raw) = value("BASANGO_CRAWLER_ROOT_PATH") {
config.paths.root = raw.into();
}
if let Some(raw) = value("BASANGO_CRAWLER_SQLITE_PATH") {
config.paths.sqlite = Some(raw.into());
}
Ok(())
}
pub(super) fn value(name: &str) -> Option<String> {
env::var(name)
.ok()
.map(|value| value.trim().to_owned())
.filter(|value| !value.is_empty())
}
fn set_string(name: &str, target: &mut String) {
if let Some(raw) = value(name) {
*target = raw;
}
}
fn set_parsed<T>(name: &str, target: &mut T) -> Result<()>
where
T: FromStr,
{
if let Some(raw) = value(name) {
*target = raw.parse().map_err(|_| {
CrawlError::Configuration(format!(
"environment variable {name} has invalid value '{raw}'"
))
})?;
}
Ok(())
}
fn parse_bool(name: &str, raw: &str) -> Result<bool> {
match raw.to_ascii_lowercase().as_str() {
"1" | "true" | "yes" | "on" => Ok(true),
"0" | "false" | "no" | "off" => Ok(false),
_ => Err(CrawlError::Configuration(format!(
"environment variable {name} must be a boolean, got '{raw}'"
))),
}
}
+55
View File
@@ -0,0 +1,55 @@
use std::time::Duration;
use serde::{Deserialize, Serialize};
#[derive(Debug, Clone, Deserialize, Serialize)]
#[serde(default, deny_unknown_fields)]
pub struct HttpClientConfig {
pub backoff: BackoffConfig,
pub follow_redirects: bool,
pub max_retries: u32,
pub respect_retry_after: bool,
pub rotate: bool,
pub timeout: u64,
pub user_agent: String,
pub verify_ssl: bool,
}
impl HttpClientConfig {
pub fn timeout(&self) -> Duration {
Duration::from_secs(self.timeout)
}
}
impl Default for HttpClientConfig {
fn default() -> Self {
Self {
backoff: BackoffConfig::default(),
follow_redirects: true,
max_retries: 3,
respect_retry_after: true,
rotate: true,
timeout: 20,
user_agent: "Basango/0.1 (+https://github.com/bernard-ng/basango)".into(),
verify_ssl: true,
}
}
}
#[derive(Debug, Clone, Copy, Deserialize, Serialize)]
#[serde(default, deny_unknown_fields)]
pub struct BackoffConfig {
pub initial: f64,
pub max: f64,
pub multiplier: f64,
}
impl Default for BackoffConfig {
fn default() -> Self {
Self {
initial: 1.0,
max: 30.0,
multiplier: 2.0,
}
}
}
+9
View File
@@ -0,0 +1,9 @@
use serde::{Deserialize, Serialize};
use url::Url;
#[derive(Debug, Clone, Default, Deserialize, Serialize)]
#[serde(default, deny_unknown_fields)]
pub struct IngestionApiConfig {
pub endpoint: Option<Url>,
pub token: String,
}
+63
View File
@@ -0,0 +1,63 @@
use std::{borrow::Cow, fs, path::Path, path::PathBuf};
use serde_json::Value;
use crate::error::{CrawlError, Result};
use super::{CrawlerConfig, environment, schema, validation};
pub(super) const BUNDLED_CONFIG: &str = include_str!("../../config/crawler.json");
pub(super) fn load(path: Option<PathBuf>) -> Result<CrawlerConfig> {
let _ = dotenvy::dotenv();
let path =
path.or_else(|| environment::value("BASANGO_CRAWLER_CONFIG_PATH").map(PathBuf::from));
let raw = read(path.as_deref())?;
let mut config = decode(&raw)?;
environment::apply(&mut config)?;
validation::validate(&config)?;
Ok(config)
}
#[cfg(test)]
pub(super) fn parse(raw: &str) -> Result<CrawlerConfig> {
let config = decode(raw)?;
validation::validate(&config)?;
Ok(config)
}
fn decode(raw: &str) -> Result<CrawlerConfig> {
let value: Value = serde_json::from_str(raw)?;
schema::validate(&value)?;
serde_json::from_value(value).map_err(Into::into)
}
fn read(path: Option<&Path>) -> Result<Cow<'static, str>> {
match path {
Some(path) => fs::read_to_string(path).map(Cow::Owned).map_err(|error| {
CrawlError::Configuration(format!("cannot read {}: {error}", path.display()))
}),
None => Ok(Cow::Borrowed(BUNDLED_CONFIG)),
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn missing_external_path_borrows_the_bundled_configuration() {
assert!(matches!(read(None).unwrap(), Cow::Borrowed(_)));
}
#[test]
fn structural_decode_allows_environment_to_complete_secrets() {
let raw = r#"{
"ingestion": { "endpoint": "https://api.example.com" },
"sources": [{ "kind": "wordpress", "id": "example", "url": "https://example.com" }]
}"#;
assert!(decode(raw).is_ok());
assert!(parse(raw).is_err());
}
}
+148
View File
@@ -0,0 +1,148 @@
//! Typed crawler configuration.
//!
//! Loading, environment overrides, structural schemas, semantic validation,
//! and each configuration area live in separate modules. Callers consume the
//! typed facade exported here.
mod environment;
mod http;
mod ingestion;
mod loader;
mod paths;
mod queue;
mod runtime;
mod schema;
mod source;
mod validation;
use std::path::PathBuf;
use serde::{Deserialize, Serialize};
use crate::{
domain::SourceId,
error::{CrawlError, Result},
};
pub use http::{BackoffConfig, HttpClientConfig};
pub use ingestion::IngestionApiConfig;
pub use paths::PathsConfig;
pub use queue::{JobRetention, QueueConfig, QueueNames};
pub use runtime::CrawlerRuntimeConfig;
pub use source::{
CommonSourceConfig, HtmlSelectors, HtmlSourceConfig, MetadataStrategy, SourceConfig,
WordPressSourceConfig,
};
#[derive(Debug, Clone, Default, Deserialize, Serialize)]
#[serde(default, deny_unknown_fields)]
pub struct CrawlerConfig {
#[serde(alias = "backend")]
pub ingestion: IngestionApiConfig,
pub http: HttpClientConfig,
pub paths: PathsConfig,
pub queue: QueueConfig,
pub runtime: CrawlerRuntimeConfig,
pub sources: Vec<SourceConfig>,
}
impl CrawlerConfig {
/// Load JSON, apply environment overrides, and validate the final value.
pub fn load(path: Option<PathBuf>) -> Result<Self> {
loader::load(path)
}
/// Validate configurations constructed by embedding applications.
pub fn validate(&self) -> Result<()> {
validation::validate(self)
}
pub fn source(&self, source_id: &SourceId) -> Result<SourceConfig> {
self.sources
.iter()
.find(|source| source.id() == source_id)
.cloned()
.ok_or_else(|| CrawlError::SourceNotFound(source_id.to_string()))
}
pub fn data_path(&self) -> PathBuf {
self.paths.data_path()
}
pub fn sqlite_path(&self) -> PathBuf {
self.paths.sqlite_path()
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn bundled_configuration_matches_the_zod_and_rust_schemas() {
let config = loader::parse(loader::BUNDLED_CONFIG).unwrap();
assert_eq!(config.queue.queues.discovery, "discovery");
assert_eq!(config.queue.queues.articles, "articles");
assert!(matches!(config.sources[0], SourceConfig::Html(_)));
}
#[test]
fn zod_schema_reports_nested_configuration_paths() {
let error = loader::parse(
r#"{
"http": { "timeout": 0 },
"sources": [{ "kind": "wordpress", "id": "example", "url": "not-a-url" }]
}"#,
)
.unwrap_err();
let message = error.to_string();
assert!(message.contains("http.timeout"), "{message}");
assert!(message.contains("sources.0.url"), "{message}");
}
#[test]
fn duplicate_source_ids_are_rejected_semantically() {
let error = loader::parse(
r#"{
"sources": [
{ "kind": "wordpress", "id": "duplicate", "url": "https://one.example" },
{ "kind": "wordpress", "id": "duplicate", "url": "https://two.example" }
]
}"#,
)
.unwrap_err();
assert!(
error
.to_string()
.contains("duplicate source id 'duplicate'")
);
}
#[test]
fn former_backend_key_remains_a_configuration_alias() {
let config = loader::parse(
r#"{
"backend": { "endpoint": "https://api.example.com", "token": "secret" },
"sources": [{ "kind": "wordpress", "id": "example", "url": "https://example.com" }]
}"#,
)
.unwrap();
assert_eq!(
config.ingestion.endpoint.unwrap().as_str(),
"https://api.example.com/"
);
}
#[test]
fn typescript_wrapper_is_not_a_supported_configuration_shape() {
let result = loader::parse(
r#"{
"crawler": {
"sources": [{ "kind": "wordpress", "id": "example", "url": "https://example.com" }]
}
}"#,
);
assert!(result.is_err());
}
}
+37
View File
@@ -0,0 +1,37 @@
use std::path::PathBuf;
use serde::{Deserialize, Serialize};
#[derive(Debug, Clone, Deserialize, Serialize)]
#[serde(default, deny_unknown_fields)]
pub struct PathsConfig {
pub root: PathBuf,
pub data: PathBuf,
pub sqlite: Option<PathBuf>,
}
impl PathsConfig {
pub(super) fn data_path(&self) -> PathBuf {
if self.data.as_os_str().is_empty() {
self.root.join("data")
} else {
self.data.clone()
}
}
pub(super) fn sqlite_path(&self) -> PathBuf {
self.sqlite
.clone()
.unwrap_or_else(|| self.data_path().join("crawler.db"))
}
}
impl Default for PathsConfig {
fn default() -> Self {
Self {
root: PathBuf::from("."),
data: PathBuf::new(),
sqlite: None,
}
}
}
+53
View File
@@ -0,0 +1,53 @@
use serde::{Deserialize, Serialize};
#[derive(Debug, Clone, Deserialize, Serialize)]
#[serde(default, deny_unknown_fields)]
pub struct QueueConfig {
pub prefix: String,
pub queues: QueueNames,
pub redis_url: String,
pub retention: JobRetention,
}
impl Default for QueueConfig {
fn default() -> Self {
Self {
prefix: "basango:crawler".into(),
queues: QueueNames::default(),
redis_url: "redis://localhost:6379/0".into(),
retention: JobRetention::default(),
}
}
}
#[derive(Debug, Clone, Deserialize, Serialize)]
#[serde(default, deny_unknown_fields)]
pub struct QueueNames {
pub discovery: String,
pub articles: String,
}
impl Default for QueueNames {
fn default() -> Self {
Self {
discovery: "discovery".into(),
articles: "articles".into(),
}
}
}
#[derive(Debug, Clone, Copy, Deserialize, Serialize)]
#[serde(default, deny_unknown_fields)]
pub struct JobRetention {
pub completed: u64,
pub failed: u64,
}
impl Default for JobRetention {
fn default() -> Self {
Self {
completed: 3_600,
failed: 86_400,
}
}
}
+19
View File
@@ -0,0 +1,19 @@
use serde::{Deserialize, Serialize};
use crate::domain::UpdateDirection;
#[derive(Debug, Clone, Deserialize, Serialize)]
#[serde(default, deny_unknown_fields)]
pub struct CrawlerRuntimeConfig {
pub direction: UpdateDirection,
pub worker_concurrency: usize,
}
impl Default for CrawlerRuntimeConfig {
fn default() -> Self {
Self {
direction: UpdateDirection::Forward,
worker_concurrency: 5,
}
}
}
+176
View File
@@ -0,0 +1,176 @@
use serde_json::Value;
use zod_rs::prelude::*;
use crate::error::{CrawlError, Result};
pub(super) fn validate(value: &Value) -> Result<()> {
let mut failures = Vec::new();
if let Err(errors) = crawler_schema().safe_parse(value) {
failures.push(errors.to_string());
}
if let Some(sources) = value.get("sources").and_then(Value::as_array) {
for (index, source) in sources.iter().enumerate() {
let result = match source.get("kind").and_then(Value::as_str) {
Some("html") => html_source_schema().safe_parse(source),
Some("wordpress") => wordpress_source_schema().safe_parse(source),
_ => source_kind_schema().safe_parse(source),
};
if let Err(mut errors) = result {
errors.prefix_path(index.to_string());
errors.prefix_path("sources".into());
failures.push(errors.to_string());
}
}
}
if failures.is_empty() {
Ok(())
} else {
Err(CrawlError::Configuration(format!(
"schema validation failed:{}",
failures.concat()
)))
}
}
fn crawler_schema() -> ObjectSchema {
object()
.optional_field("backend", ingestion_schema())
.optional_field("ingestion", ingestion_schema())
.optional_field("http", http_schema())
.optional_field("paths", paths_schema())
.optional_field("queue", queue_schema())
.optional_field("runtime", runtime_schema())
.field("sources", array(object()).min(1))
.strict()
}
fn ingestion_schema() -> ObjectSchema {
object()
.optional_field("endpoint", string().url())
.optional_field("token", string())
.strict()
}
fn paths_schema() -> ObjectSchema {
object()
.optional_field("root", string())
.optional_field("data", string())
.optional_field("sqlite", string())
.strict()
}
fn queue_schema() -> ObjectSchema {
object()
.optional_field("prefix", non_blank_string())
.optional_field("queues", queue_names_schema())
.optional_field("redis_url", string().regex(r"^rediss?://"))
.optional_field("retention", retention_schema())
.strict()
}
fn queue_names_schema() -> ObjectSchema {
object()
.optional_field("discovery", non_blank_string())
.optional_field("articles", non_blank_string())
.strict()
}
fn retention_schema() -> ObjectSchema {
object()
.optional_field("completed", number().int().nonnegative())
.optional_field("failed", number().int().nonnegative())
.strict()
}
fn http_schema() -> ObjectSchema {
object()
.optional_field("backoff", backoff_schema())
.optional_field("follow_redirects", boolean())
.optional_field("max_retries", number().int().nonnegative())
.optional_field("respect_retry_after", boolean())
.optional_field("rotate", boolean())
.optional_field("timeout", number().int().positive())
.optional_field("user_agent", non_blank_string())
.optional_field("verify_ssl", boolean())
.strict()
}
fn backoff_schema() -> ObjectSchema {
object()
.optional_field("initial", number().positive().finite())
.optional_field("max", number().positive().finite())
.optional_field("multiplier", number().positive().finite())
.strict()
}
fn runtime_schema() -> ObjectSchema {
object()
.optional_field("direction", direction_schema())
.optional_field("worker_concurrency", number().int().positive())
.strict()
}
fn html_source_schema() -> ObjectSchema {
source_base_schema("html")
.field("pagination_template", non_blank_string())
.field("selectors", selectors_schema())
.optional_field("fetch_details", boolean())
.strict()
}
fn wordpress_source_schema() -> ObjectSchema {
source_base_schema("wordpress")
.optional_field("metadata_strategy", metadata_strategy_schema())
.strict()
}
fn source_kind_schema() -> ObjectSchema {
object().field(
"kind",
union()
.variant(literal("html"))
.variant(literal("wordpress")),
)
}
fn source_base_schema(kind: &'static str) -> ObjectSchema {
object()
.field("kind", literal(kind))
.field("id", non_blank_string())
.field("url", string().url())
.optional_field("date_format", non_blank_string())
.optional_field("rate_limit", boolean())
}
fn selectors_schema() -> ObjectSchema {
object()
.field("body", non_blank_string())
.optional_field("categories", non_blank_string())
.field("date", non_blank_string())
.field("link", non_blank_string())
.field("list", non_blank_string())
.field("title", non_blank_string())
.optional_field("pagination", non_blank_string())
.strict()
}
fn direction_schema() -> UnionSchema<String> {
union()
.variant(literal("forward"))
.variant(literal("backward"))
}
fn metadata_strategy_schema() -> UnionSchema<String> {
union()
.variant(literal("auto"))
.variant(literal("yoast"))
.variant(literal("rest"))
.variant(literal("fetch"))
.variant(literal("none"))
}
fn non_blank_string() -> StringSchema {
string().min(1).regex(r"\S")
}
+129
View File
@@ -0,0 +1,129 @@
use serde::{Deserialize, Serialize};
use url::Url;
use crate::domain::SourceId;
#[derive(Debug, Clone, Deserialize, Serialize)]
pub struct CommonSourceConfig {
#[serde(default = "default_date_format")]
pub date_format: String,
pub id: SourceId,
#[serde(default)]
pub rate_limit: bool,
pub url: Url,
}
impl Default for CommonSourceConfig {
fn default() -> Self {
Self {
date_format: default_date_format(),
id: SourceId::new("unnamed").expect("static source id is valid"),
rate_limit: false,
url: Url::parse("http://localhost").expect("static URL is valid"),
}
}
}
#[derive(Debug, Clone, Deserialize, Serialize)]
pub struct HtmlSourceConfig {
#[serde(flatten)]
pub common: CommonSourceConfig,
#[serde(default)]
pub fetch_details: bool,
pub pagination_template: String,
pub selectors: HtmlSelectors,
}
#[derive(Debug, Clone, Deserialize, Serialize)]
#[serde(deny_unknown_fields)]
pub struct HtmlSelectors {
pub body: String,
#[serde(default)]
pub categories: Option<String>,
pub date: String,
pub link: String,
pub list: String,
pub title: String,
#[serde(default = "default_pagination_selector")]
pub pagination: String,
}
#[derive(Debug, Clone, Deserialize, Serialize)]
pub struct WordPressSourceConfig {
#[serde(flatten)]
pub common: CommonSourceConfig,
#[serde(default)]
pub metadata_strategy: MetadataStrategy,
}
#[derive(Debug, Clone, Copy, Default, Deserialize, Serialize)]
#[serde(rename_all = "lowercase")]
pub enum MetadataStrategy {
#[default]
Auto,
Yoast,
Rest,
Fetch,
None,
}
#[derive(Debug, Clone, Deserialize, Serialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum SourceConfig {
Html(HtmlSourceConfig),
#[serde(rename = "wordpress")]
WordPress(WordPressSourceConfig),
}
impl SourceConfig {
pub fn id(&self) -> &SourceId {
&self.common().id
}
pub fn url(&self) -> &Url {
&self.common().url
}
pub fn common(&self) -> &CommonSourceConfig {
match self {
Self::Html(source) => &source.common,
Self::WordPress(source) => &source.common,
}
}
}
fn default_pagination_selector() -> String {
"ul.pagination > li a".into()
}
fn default_date_format() -> String {
"yyyy-LL-dd HH:mm".into()
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn source_config_deserializes_directly_into_its_variant() {
let json = r#"{
"kind": "html",
"id": "example",
"url": "https://example.com",
"pagination_template": "news",
"selectors": {
"body": ".body",
"date": "time",
"link": "a",
"list": ".article",
"title": "h1"
}
}"#;
let source: SourceConfig = serde_json::from_str(json).unwrap();
let SourceConfig::Html(source) = source else {
panic!("expected HTML source");
};
assert_eq!(source.selectors.list, ".article");
assert_eq!(source.common.id.as_str(), "example");
}
}
+36
View File
@@ -0,0 +1,36 @@
use std::collections::HashSet;
use crate::error::{CrawlError, Result};
use super::{CrawlerConfig, schema};
pub(super) fn validate(config: &CrawlerConfig) -> Result<()> {
schema::validate(&serde_json::to_value(config)?)?;
if config.queue.queues.discovery == config.queue.queues.articles {
return Err(CrawlError::Configuration(
"discovery and article queue names must be distinct".into(),
));
}
if config.ingestion.endpoint.is_some() && config.ingestion.token.trim().is_empty() {
return Err(CrawlError::Configuration(
"ingestion.token is required when ingestion.endpoint is configured".into(),
));
}
if config.http.backoff.max < config.http.backoff.initial {
return Err(CrawlError::Configuration(
"http.backoff.max must be greater than or equal to http.backoff.initial".into(),
));
}
let mut ids = HashSet::new();
for source in &config.sources {
if !ids.insert(source.id()) {
return Err(CrawlError::Configuration(format!(
"duplicate source id '{}'",
source.id()
)));
}
}
Ok(())
}
+69
View File
@@ -0,0 +1,69 @@
//! Small public facade for embedding the crawler in another Rust program.
use std::path::PathBuf;
use crate::{
config::CrawlerConfig,
domain::{CrawlRequest, SourceId},
error::Result,
execution::{
CrawlReport, DiscoverJob, JobQueue, Runtime, crawl_now, forward_pending, run_worker,
},
};
/// A configured crawler with reusable HTTP connections.
#[derive(Clone)]
pub struct Crawler {
runtime: Runtime,
}
impl Crawler {
pub fn new(config: CrawlerConfig) -> Result<Self> {
Ok(Self {
runtime: Runtime::new(config)?,
})
}
/// Use an environment-selected config file or the bundled default, then
/// apply environment overrides.
pub fn from_environment() -> Result<Self> {
Self::new(CrawlerConfig::load(None)?)
}
/// Load one explicit JSON config file, then apply environment overrides.
pub fn from_config_file(path: impl Into<PathBuf>) -> Result<Self> {
Self::new(CrawlerConfig::load(Some(path.into()))?)
}
pub fn config(&self) -> &CrawlerConfig {
&self.runtime.config
}
/// Crawl now, streaming collected drafts into the durable outbox.
pub async fn crawl(&self, request: CrawlRequest) -> Result<CrawlReport> {
crawl_now(&self.runtime, request).await
}
/// Schedule source discovery in BullMQ.
pub async fn schedule(&self, request: CrawlRequest) -> Result<String> {
self.runtime.config.source(&request.source_id)?;
JobQueue::connect(&self.runtime.config.queue)
.await?
.enqueue_discovery(DiscoverJob { request })
.await
}
/// Deliver pending and retryable outbox entries.
pub async fn deliver_pending(
&self,
source: Option<&SourceId>,
limit: usize,
) -> Result<CrawlReport> {
forward_pending(&self.runtime, source, limit).await
}
/// Run BullMQ consumers until the process receives Ctrl-C.
pub async fn work(&self, queues: Vec<String>, concurrency: usize) -> Result<()> {
run_worker(self.runtime.clone(), queues, concurrency).await
}
}
+276
View File
@@ -0,0 +1,276 @@
//! Domain types: the vocabulary of the crawler.
//!
//! Domain values describe *what* Basango works with. They deliberately do not
//! know how HTTP, Redis, SQLite, or the CLI work. This dependency direction is
//! what lets the same types move through synchronous and queued execution.
use std::{fmt, str::FromStr};
use chrono::{DateTime, NaiveDate, TimeZone, Utc};
use serde::{Deserialize, Serialize};
use url::Url;
use crate::error::{CrawlError, Result};
// --- Source identity -------------------------------------------------------
/// A validated source identifier.
///
/// Using a newtype prevents an arbitrary or empty `String` from being passed
/// wherever the crawler expects the identity of a configured source.
#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize)]
#[serde(transparent)]
pub struct SourceId(String);
impl SourceId {
pub fn new(value: impl Into<String>) -> Result<Self> {
let value = value.into();
let normalized = value.trim();
if normalized.is_empty() {
return Err(CrawlError::Configuration(
"source id cannot be empty".into(),
));
}
Ok(Self(normalized.to_owned()))
}
pub fn as_str(&self) -> &str {
&self.0
}
}
impl fmt::Display for SourceId {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter.write_str(&self.0)
}
}
impl FromStr for SourceId {
type Err = CrawlError;
fn from_str(value: &str) -> Result<Self> {
Self::new(value)
}
}
impl AsRef<str> for SourceId {
fn as_ref(&self) -> &str {
self.as_str()
}
}
impl<'de> Deserialize<'de> for SourceId {
fn deserialize<D>(deserializer: D) -> std::result::Result<Self, D::Error>
where
D: serde::Deserializer<'de>,
{
let value = String::deserialize(deserializer)?;
Self::new(value).map_err(serde::de::Error::custom)
}
}
// --- Articles -------------------------------------------------------------
/// Optional metadata discovered from Open Graph or WordPress fields.
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct ArticleMetadata {
pub url: Option<Url>,
pub title: Option<String>,
pub author: Option<String>,
pub description: Option<String>,
pub image: Option<Url>,
pub published_at: Option<String>,
pub updated_at: Option<String>,
}
impl ArticleMetadata {
/// Empty metadata is represented as `None` instead of an object full of
/// `null` values. That makes absence explicit to downstream code.
pub fn is_empty(&self) -> bool {
self.url.is_none()
&& self.title.is_none()
&& self.author.is_none()
&& self.description.is_none()
&& self.image.is_none()
&& self.published_at.is_none()
&& self.updated_at.is_none()
}
}
/// A source crawler's output before normalization and hashing.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ArticleDraft {
pub title: String,
pub body: String,
pub link: Url,
pub source_id: SourceId,
pub categories: Vec<String>,
pub metadata: Option<ArticleMetadata>,
pub published_at: DateTime<Utc>,
}
/// The validated representation persisted in the outbox and sent to the API.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct Article {
pub hash: String,
pub title: String,
pub body: String,
pub link: Url,
pub source_id: SourceId,
#[serde(default)]
pub categories: Vec<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub metadata: Option<ArticleMetadata>,
pub published_at: DateTime<Utc>,
}
// --- Ranges and crawl options --------------------------------------------
/// Inclusive page boundaries. HTML sources may start at page zero, whereas
/// WordPress starts at page one, so zero is a valid value here.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub struct PageRange {
pub start: u32,
pub end: u32,
}
impl PageRange {
pub fn new(start: u32, end: u32) -> Result<Self> {
if end < start {
return Err(CrawlError::InvalidRange(format!(
"end page {end} is before start page {start}"
)));
}
Ok(Self { start, end })
}
/// Parse the CLI representation `start:end`.
pub fn parse(spec: &str) -> Result<Self> {
let (start, end) = split_range(spec, "page")?;
Self::new(
start
.parse()
.map_err(|_| CrawlError::InvalidRange(format!("invalid start page '{start}'")))?,
end.parse()
.map_err(|_| CrawlError::InvalidRange(format!("invalid end page '{end}'")))?,
)
}
}
impl fmt::Display for PageRange {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(formatter, "{}:{}", self.start, self.end)
}
}
/// Inclusive UTC time boundaries used to filter articles.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub struct DateRange {
pub start: DateTime<Utc>,
pub end: DateTime<Utc>,
}
impl DateRange {
pub fn new(start: DateTime<Utc>, end: DateTime<Utc>) -> Result<Self> {
if end < start {
return Err(CrawlError::InvalidRange(
"end date must be on or after start date".into(),
));
}
Ok(Self { start, end })
}
/// Parse `YYYY-MM-DD:YYYY-MM-DD`. The end date is expanded to the last
/// nanosecond of that day so the range behaves as users expect.
pub fn parse(spec: &str) -> Result<Self> {
let (start, end) = split_range(spec, "date")?;
let start = parse_date(start)?
.and_hms_opt(0, 0, 0)
.expect("midnight is valid");
let end = parse_date(end)?
.and_hms_nano_opt(23, 59, 59, 999_999_999)
.expect("end of day is valid");
Self::new(Utc.from_utc_datetime(&start), Utc.from_utc_datetime(&end))
}
pub fn contains(&self, timestamp: DateTime<Utc>) -> bool {
self.start <= timestamp && timestamp <= self.end
}
/// Crawlers receive newest-first listings. Once an article is older than
/// `start`, all following items are normally older too and crawling can stop.
pub fn is_older_than_range(&self, timestamp: DateTime<Utc>) -> bool {
timestamp < self.start
}
}
impl fmt::Display for DateRange {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(
formatter,
"{}:{}",
self.start.format("%Y-%m-%d"),
self.end.format("%Y-%m-%d")
)
}
}
/// One request to crawl a configured source.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct CrawlRequest {
pub source_id: SourceId,
pub page_range: Option<PageRange>,
pub date_range: Option<DateRange>,
pub category: Option<String>,
}
/// Crawling direction used when the backend supplies an update boundary.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum UpdateDirection {
Backward,
#[default]
Forward,
}
// --- Helpers --------------------------------------------------------------
fn split_range<'a>(spec: &'a str, kind: &str) -> Result<(&'a str, &'a str)> {
spec.split_once(':').ok_or_else(|| {
CrawlError::InvalidRange(format!("invalid {kind} range '{spec}'; expected start:end"))
})
}
fn parse_date(value: &str) -> Result<NaiveDate> {
NaiveDate::parse_from_str(value, "%Y-%m-%d")
.map_err(|_| CrawlError::InvalidRange(format!("invalid date '{value}'; use YYYY-MM-DD")))
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn page_range_rejects_reversed_bounds() {
assert!(PageRange::parse("5:2").is_err());
}
#[test]
fn timestamp_range_includes_the_whole_end_day() {
let range = DateRange::parse("2025-01-01:2025-01-02").unwrap();
let end_of_day = DateTime::parse_from_rfc3339("2025-01-02T23:59:59Z")
.unwrap()
.with_timezone(&Utc);
assert!(range.contains(end_of_day));
}
#[test]
fn source_id_is_trimmed_and_cannot_be_empty() {
assert_eq!(SourceId::new(" example ").unwrap().as_str(), "example");
assert!(SourceId::new(" ").is_err());
assert!(serde_json::from_str::<SourceId>(r#""""#).is_err());
}
}
+62
View File
@@ -0,0 +1,62 @@
//! One error vocabulary for the crawler.
//!
//! `thiserror` removes repetitive `Display` and `From` implementations while
//! keeping errors typed. `anyhow` is only used at the executable boundary,
//! where reporting matters more than programmatic recovery.
use thiserror::Error;
/// Convenience alias used throughout the library.
pub type Result<T, E = CrawlError> = std::result::Result<T, E>;
#[derive(Debug, Error)]
pub enum CrawlError {
#[error("configuration error: {0}")]
Configuration(String),
#[error("source '{0}' was not found")]
SourceNotFound(String),
#[error("invalid source selectors: {0}")]
InvalidSourceSelectors(String),
#[error("invalid article: {0}")]
InvalidArticle(String),
#[error("article at {url} is outside the requested date range")]
ArticleOutOfDateRange { url: String },
#[error("invalid range: {0}")]
InvalidRange(String),
#[error("HTTP request failed: {0}")]
HttpTransport(#[from] reqwest::Error),
#[error("HTTP {status} from {url}: {body}")]
HttpStatus {
status: u16,
url: String,
body: String,
},
#[error("SQLite outbox error: {0}")]
Sqlite(#[from] rusqlite::Error),
#[error("SQLite outbox became unavailable after a task panicked")]
OutboxLockPoisoned,
#[error("BullMQ error: {0}")]
BullMq(#[from] bullmq::Error),
#[error("JSON error: {0}")]
Json(#[from] serde_json::Error),
#[error("I/O error: {0}")]
Io(#[from] std::io::Error),
#[error("URL error: {0}")]
Url(#[from] url::ParseError),
#[error("queue error: {0}")]
Queue(String),
}
+94
View File
@@ -0,0 +1,94 @@
//! Application orchestration.
//!
//! Source modules know how to collect; article modules know how to persist and
//! deliver. Execution modules coordinate those capabilities for each command.
mod queue;
mod sync;
mod worker;
pub(crate) use queue::{DiscoverJob, FetchJob, JobQueue};
pub use sync::CrawlReport;
pub(crate) use sync::{crawl_now, forward_pending};
pub(crate) use worker::run_worker;
use std::sync::Arc;
use chrono::{DateTime, Utc};
use serde::Deserialize;
use crate::{
articles::endpoint_url,
config::CrawlerConfig,
domain::{CrawlRequest, DateRange, UpdateDirection},
error::Result,
http::HttpClient,
};
/// Shared, immutable dependencies are placed in `Arc` so queued jobs can own a
/// cheap reference while running concurrently.
#[derive(Clone)]
pub(crate) struct Runtime {
pub config: Arc<CrawlerConfig>,
pub http: HttpClient,
}
impl Runtime {
pub fn new(config: CrawlerConfig) -> Result<Self> {
config.validate()?;
let http = HttpClient::new(&config.http)?;
Ok(Self {
config: Arc::new(config),
http,
})
}
/// 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.
pub async fn resolve_date_range(&self, request: &mut CrawlRequest) {
if request.date_range.is_some() {
return;
}
let Some(base) = &self.config.ingestion.endpoint else {
return;
};
let Ok(endpoint) = endpoint_url(base, "ingest/sources/publication-bounds") else {
return;
};
let headers = [("Authorization", self.config.ingestion.token.as_str())];
let payload = serde_json::json!({ "name": request.source_id.as_str() });
let response = match self.http.post_json(&endpoint, &headers, &payload).await {
Ok(response) if response.is_success() => response,
Ok(response) => {
tracing::warn!(status = %response.status, "publication-bound lookup failed");
return;
}
Err(error) => {
tracing::warn!(%error, "publication-bound lookup failed");
return;
}
};
let dates: SourcePublicationBounds = match response.json() {
Ok(dates) => dates,
Err(error) => {
tracing::warn!(%error, "ingestion API returned invalid publication bounds");
return;
}
};
let now = Utc::now();
let start = match self.config.runtime.direction {
UpdateDirection::Forward => dates.latest.unwrap_or(dates.earliest),
UpdateDirection::Backward => dates.earliest,
};
request.date_range = DateRange::new(start, now).ok();
}
}
#[derive(Debug, Deserialize)]
#[serde(rename_all = "camelCase")]
struct SourcePublicationBounds {
earliest: DateTime<Utc>,
latest: Option<DateTime<Utc>>,
}
+153
View File
@@ -0,0 +1,153 @@
//! BullMQ-backed jobs used by `schedule` and `worker`.
//!
//! BullMQ owns queue state transitions, retry backoff, job locks, lock renewal,
//! stalled-job recovery, and retention. This module only defines Basango's
//! typed payloads and translates crawler configuration into BullMQ options.
use std::time::Duration;
use bullmq::options::RedisConnectionOptions;
use bullmq::types::{BackoffStrategy, KeepJobs, RemoveOnFinish};
use bullmq::{Queue, QueueOptions};
use serde::{Deserialize, Serialize};
use crate::{
config::QueueConfig,
domain::CrawlRequest,
error::{CrawlError, Result},
sources::ArticleSeed,
};
const JOB_ATTEMPTS: u32 = 3;
const RETRY_DELAY_MS: u64 = 1_000;
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DiscoverJob {
pub request: CrawlRequest,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct FetchJob {
pub request: CrawlRequest,
pub article: ArticleSeed,
}
/// Producer-side access to the two crawler queues.
pub struct JobQueue {
discovery: Queue,
articles: Queue,
config: QueueConfig,
}
impl JobQueue {
pub async fn connect(config: &QueueConfig) -> Result<Self> {
let options = queue_options(config);
let (discovery, articles) = tokio::try_join!(
Queue::with_options(&config.queues.discovery, options.clone()),
Queue::with_options(&config.queues.articles, options),
)?;
Ok(Self {
discovery,
articles,
config: config.clone(),
})
}
pub fn names(&self) -> [&str; 2] {
[
self.config.queues.discovery.as_str(),
self.config.queues.articles.as_str(),
]
}
pub fn validate_names(&self, names: &[String]) -> Result<()> {
let valid = self.names();
for name in names {
if !valid.contains(&name.as_str()) {
return Err(CrawlError::Queue(format!(
"unknown queue '{name}'; expected {} or {}",
valid[0], valid[1]
)));
}
}
Ok(())
}
pub async fn enqueue_discovery(&self, job: DiscoverJob) -> Result<String> {
let id = stable_job_id("discover", &job)?;
let queued = self
.discovery
.add("discover-source", job)
.job_id(&id)
.attempts(JOB_ATTEMPTS)
.backoff(BackoffStrategy::Exponential(RETRY_DELAY_MS))
.remove_on_complete(retention(self.config.retention.completed))
.remove_on_fail(retention(self.config.retention.failed))
.await?;
Ok(queued.id().to_owned())
}
pub async fn enqueue_article(&self, job: FetchJob) -> Result<String> {
let identity = (&job.request.source_id, &job.article.url);
let id = stable_job_id("article", &identity)?;
let queued = self
.articles
.add("fetch-article", job)
.job_id(&id)
.attempts(JOB_ATTEMPTS)
.backoff(BackoffStrategy::Exponential(RETRY_DELAY_MS))
.remove_on_complete(retention(self.config.retention.completed))
.remove_on_fail(retention(self.config.retention.failed))
.await?;
Ok(queued.id().to_owned())
}
}
pub fn queue_options(config: &QueueConfig) -> QueueOptions {
QueueOptions::new()
.connection(redis_options(config))
.prefix(config.prefix.clone())
}
pub fn redis_options(config: &QueueConfig) -> RedisConnectionOptions {
RedisConnectionOptions {
url: config.redis_url.clone(),
..RedisConnectionOptions::default()
}
}
fn retention(seconds: u64) -> RemoveOnFinish {
if seconds == 0 {
RemoveOnFinish::Bool(true)
} else {
RemoveOnFinish::Options(KeepJobs {
age: Some(Duration::from_secs(seconds).as_millis() as u64),
count: None,
limit: Some(1_000),
})
}
}
fn stable_job_id(prefix: &str, value: &impl Serialize) -> Result<String> {
let bytes = serde_json::to_vec(value)?;
Ok(format!("{prefix}-{:x}", md5::compute(bytes)))
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn stable_ids_are_deterministic() {
let value = ("example", "https://example.com/story");
assert_eq!(
stable_job_id("article", &value).unwrap(),
stable_job_id("article", &value).unwrap()
);
}
#[test]
fn zero_retention_removes_jobs_immediately() {
assert!(matches!(retention(0), RemoveOnFinish::Bool(true)));
}
}
+184
View File
@@ -0,0 +1,184 @@
//! Direct crawl and outbox delivery workflows.
use std::time::Duration as StdDuration;
use std::time::Instant;
use chrono::Duration;
use tokio::time::{Duration as TokioDuration, interval};
use uuid::Uuid;
use crate::{
articles::{ArticleIngestionClient, DeliveryResult, IngestStatus, Outbox, ingest},
domain::{CrawlRequest, SourceId},
error::{CrawlError, Result},
execution::Runtime,
sources::SourceAdapter,
telemetry::{RunMetrics, RunReporter},
};
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct CrawlReport {
pub collected: usize,
pub stored: usize,
pub delivered: usize,
pub failed: usize,
}
/// Run one source from listing discovery through article ingestion.
pub async fn crawl_now(runtime: &Runtime, mut request: CrawlRequest) -> Result<CrawlReport> {
let reporter = RunReporter::new(
&runtime.config.ingestion,
runtime.http.clone(),
request.source_id.as_str(),
);
let heartbeat_reporter = reporter.clone();
let heartbeat_task = tokio::spawn(async move {
let mut ticker = interval(TokioDuration::from_secs(15));
loop {
ticker.tick().await;
heartbeat_reporter.heartbeat().await;
}
});
let result = crawl_with_reporter(runtime, &mut request, &reporter).await;
heartbeat_task.abort();
result
}
async fn crawl_with_reporter(
runtime: &Runtime,
request: &mut CrawlRequest,
reporter: &RunReporter,
) -> Result<CrawlReport> {
let started_at = Instant::now();
reporter.preparing().await;
let mut report = CrawlReport::default();
let outcome: Result<()> = async {
let source = runtime.config.source(&request.source_id)?;
runtime.resolve_date_range(request).await;
let adapter = SourceAdapter::new(source, runtime.http.clone());
let outbox = Outbox::open(&runtime.config.sqlite_path(), true)?;
let ingestion =
ArticleIngestionClient::new(&runtime.config.ingestion, runtime.http.clone())?;
reporter.started().await;
let mut drafts = adapter.stream(request.clone());
while let Some(item) = drafts.recv().await {
let draft = item?;
report.collected += 1;
match ingest(draft, &outbox, ingestion.as_ref()).await {
Ok((_, status)) => {
report.stored += 1;
if matches!(status, IngestStatus::Forwarded | IngestStatus::AlreadyForwarded) {
report.delivered += 1;
}
if status == IngestStatus::DeliveryFailed {
report.failed += 1;
}
}
Err(error) => {
report.failed += 1;
tracing::error!(%error, source = %request.source_id, "article ingestion failed");
}
}
reporter.progress((&report).into()).await;
}
Ok(())
}
.await;
if let Err(error) = outcome {
reporter
.failed(
(&report).into(),
elapsed_millis(started_at),
error.to_string(),
)
.await;
return Err(error);
}
reporter
.completed((&report).into(), elapsed_millis(started_at))
.await;
Ok(report)
}
/// Claim and deliver pending/failed outbox rows.
pub async fn forward_pending(
runtime: &Runtime,
source_id: Option<&SourceId>,
limit: usize,
) -> Result<CrawlReport> {
let outbox_path = runtime.config.sqlite_path();
if !Outbox::exists(&outbox_path) {
return Err(CrawlError::Configuration(format!(
"SQLite outbox does not exist: {}",
outbox_path.display()
)));
}
let ingestion = ArticleIngestionClient::new(&runtime.config.ingestion, runtime.http.clone())?
.ok_or_else(|| {
CrawlError::Configuration(
"delivery requires BASANGO_API_CRAWLER_ENDPOINT or ingestion.endpoint".into(),
)
})?;
let claim_id = format!("{}:{}", std::process::id(), Uuid::now_v7());
let outbox = Outbox::open(&outbox_path, false)?;
let articles = outbox.claim(
&claim_id,
source_id.map(SourceId::as_str),
limit,
Duration::from_std(StdDuration::from_secs(15 * 60))
.expect("15 minutes fits Chrono's duration"),
)?;
let mut report = CrawlReport {
collected: articles.len(),
stored: articles.len(),
..CrawlReport::default()
};
for record in articles {
match ingestion.deliver(&record.article).await {
DeliveryResult::Delivered { .. } => {
outbox.mark_forwarded(&record.article.hash)?;
report.delivered += 1;
}
DeliveryResult::Failed {
retryable, message, ..
} => {
outbox.mark_failed(&record.article.hash, &message, retryable)?;
report.failed += 1;
}
}
}
Ok(report)
}
impl From<&CrawlReport> for RunMetrics {
fn from(report: &CrawlReport) -> Self {
Self {
articles_discovered: report.collected,
articles_persisted: report.stored,
articles_delivered: report.delivered,
articles_failed: report.failed,
}
}
}
fn elapsed_millis(started_at: Instant) -> u64 {
started_at.elapsed().as_millis().min(u128::from(u64::MAX)) as u64
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn summary_default_starts_at_zero() {
assert_eq!(CrawlReport::default().collected, 0);
}
}
+184
View File
@@ -0,0 +1,184 @@
//! BullMQ worker orchestration.
use std::sync::Arc;
use bullmq::worker::WorkerEvent;
use bullmq::{Job, Worker, WorkerOptions};
use serde_json::{Value, json};
use tokio::sync::Semaphore;
use tokio::time::{Duration, interval};
use crate::{
articles::{ArticleIngestionClient, Outbox, ingest},
error::{CrawlError, Result},
execution::{DiscoverJob, FetchJob, JobQueue, Runtime},
sources::SourceAdapter,
telemetry::AgentReporter,
};
pub async fn run_worker(
runtime: Runtime,
queue_names: Vec<String>,
concurrency: usize,
) -> Result<()> {
let jobs = Arc::new(JobQueue::connect(&runtime.config.queue).await?);
let queue_names = if queue_names.is_empty() {
jobs.names().into_iter().map(str::to_owned).collect()
} else {
jobs.validate_names(&queue_names)?;
queue_names
};
let outbox = Outbox::open(&runtime.config.sqlite_path(), true)?;
let ingestion = ArticleIngestionClient::new(&runtime.config.ingestion, runtime.http.clone())?;
let permits = Arc::new(Semaphore::new(concurrency.max(1)));
let mut workers = Vec::with_capacity(queue_names.len());
for queue_name in &queue_names {
let worker_runtime = runtime.clone();
let jobs = jobs.clone();
let outbox = outbox.clone();
let ingestion = ingestion.clone();
let permits = permits.clone();
let processor = move |job: Job, _cancellation| {
let runtime = worker_runtime.clone();
let jobs = jobs.clone();
let outbox = outbox.clone();
let ingestion = ingestion.clone();
let permits = permits.clone();
async move {
let _permit = permits.acquire_owned().await.map_err(|_| {
bullmq::Error::ProcessingError("worker concurrency gate closed".into())
})?;
process_job(&runtime, &jobs, &outbox, ingestion.as_ref(), job).await
}
};
let options = WorkerOptions::new()
.connection(super::queue::redis_options(&runtime.config.queue))
.prefix(runtime.config.queue.prefix.clone())
.name("crawler")
// The shared semaphore is the authoritative process-wide limit.
// This value lets either queue use the full capacity while idle.
.concurrency(concurrency.max(1));
let worker = Arc::new(Worker::with_options(queue_name, processor, options).await?);
tokio::spawn(log_worker_events(worker.clone()));
workers.push(worker);
}
let heartbeat_reporter = AgentReporter::new(&runtime.config.ingestion, runtime.http.clone());
let heartbeat_task = tokio::spawn(async move {
let mut ticker = interval(Duration::from_secs(15));
loop {
ticker.tick().await;
heartbeat_reporter.heartbeat().await;
}
});
tracing::info!(?queue_names, concurrency, "BullMQ crawler worker started");
tokio::signal::ctrl_c().await.map_err(CrawlError::Io)?;
heartbeat_task.abort();
tracing::info!("shutdown requested; draining BullMQ workers");
for worker in &workers {
worker.close(30_000).await?;
}
Ok(())
}
async fn log_worker_events(worker: Arc<Worker>) {
while let Some(event) = worker.next_event().await {
match event {
WorkerEvent::Failed { job_id, error } => {
tracing::error!(job_id, %error, "BullMQ job failed");
}
WorkerEvent::Error(error) => tracing::error!(%error, "BullMQ worker error"),
WorkerEvent::Stalled { job_id } => {
tracing::warn!(job_id, "BullMQ recovered a stalled job");
}
WorkerEvent::Completed { job_id, .. } => {
tracing::debug!(job_id, "BullMQ job completed");
}
WorkerEvent::Closed => break,
_ => {}
}
}
}
async fn process_job(
runtime: &Runtime,
jobs: &JobQueue,
outbox: &Outbox,
ingestion: Option<&ArticleIngestionClient>,
job: Job,
) -> bullmq::Result<Value> {
match job.name() {
"discover-source" => {
let payload: DiscoverJob = serde_json::from_value(job.data().clone())?;
process_discovery(runtime, jobs, payload)
.await
.map(|count| json!({ "articlesQueued": count }))
.map_err(processing_error)
}
"fetch-article" => {
let payload: FetchJob = serde_json::from_value(job.data().clone())?;
process_article(runtime, outbox, ingestion, payload)
.await
.map(|()| Value::Null)
.map_err(processing_error)
}
name => Err(bullmq::Error::Unrecoverable(format!(
"unknown crawler job '{name}'"
))),
}
}
async fn process_discovery(
runtime: &Runtime,
jobs: &JobQueue,
payload: DiscoverJob,
) -> Result<usize> {
let mut request = payload.request;
runtime.resolve_date_range(&mut request).await;
let source = runtime.config.source(&request.source_id)?;
let mut adapter = SourceAdapter::new(source, runtime.http.clone());
let articles = adapter.discover(&request).await?;
let count = articles.len();
for article in articles {
jobs.enqueue_article(FetchJob {
request: request.clone(),
article,
})
.await?;
}
tracing::info!(source = %request.source_id, count, "discovery job queued articles");
Ok(count)
}
async fn process_article(
runtime: &Runtime,
outbox: &Outbox,
ingestion: Option<&ArticleIngestionClient>,
payload: FetchJob,
) -> Result<()> {
let source = runtime.config.source(&payload.request.source_id)?;
let mut adapter = SourceAdapter::new(source, runtime.http.clone());
let draft = match adapter.collect(&payload.article, &payload.request).await {
Ok(draft) => draft,
// These skips are deterministic and should count as successful jobs.
Err(CrawlError::InvalidArticle(message)) => {
tracing::info!(%message, url = %payload.article.url, "skipping invalid article");
return Ok(());
}
Err(CrawlError::ArticleOutOfDateRange { .. }) => {
tracing::info!(url = %payload.article.url, "skipping out-of-range article");
return Ok(());
}
Err(error) => return Err(error),
};
let (_, status) = ingest(draft, outbox, ingestion).await?;
tracing::info!(url = %payload.article.url, ?status, "article job completed");
Ok(())
}
fn processing_error(error: CrawlError) -> bullmq::Error {
bullmq::Error::ProcessingError(error.to_string())
}
+14
View File
@@ -0,0 +1,14 @@
//! HTTP infrastructure shared by source crawlers and backend forwarding.
//!
//! This module hides retries, timeouts, user-agent selection, and status
//! handling behind a small client. Callers focus on their protocol instead of
//! rebuilding transport policy for every request.
mod client;
mod open_graph;
mod user_agent;
pub use client::{HttpClient, HttpResponse};
pub use open_graph::{
consume_html as consume_open_graph_html, consume_url as consume_open_graph_url,
};
+234
View File
@@ -0,0 +1,234 @@
//! Retrying asynchronous HTTP client.
use std::time::{Duration, SystemTime};
use rand::Rng;
use reqwest::{
Method, StatusCode,
header::{HeaderMap, HeaderName, HeaderValue, RETRY_AFTER, USER_AGENT},
redirect::Policy,
};
use serde::{Serialize, de::DeserializeOwned};
use tokio::time::sleep;
use url::Url;
use crate::{
config::HttpClientConfig,
error::{CrawlError, Result},
};
use super::user_agent;
/// An owned response keeps callers independent of Reqwest's streaming body.
/// Crawled pages are bounded article/listing documents, so buffering is a
/// reasonable and much simpler trade-off for this application.
#[derive(Debug, Clone)]
pub struct HttpResponse {
pub status: StatusCode,
pub headers: HeaderMap,
pub url: Url,
body: Vec<u8>,
}
impl HttpResponse {
pub fn is_success(&self) -> bool {
self.status.is_success()
}
pub fn text(&self) -> Result<String> {
// News archives occasionally contain legacy bytes despite claiming
// UTF-8. Lossy decoding preserves the page's usable text instead of
// rejecting an otherwise crawlable article.
Ok(String::from_utf8_lossy(&self.body).into_owned())
}
pub fn json<T: DeserializeOwned>(&self) -> Result<T> {
Ok(serde_json::from_slice(&self.body)?)
}
/// Convert non-2xx statuses into the application's typed error.
pub fn require_success(self) -> Result<Self> {
if self.is_success() {
return Ok(self);
}
Err(CrawlError::HttpStatus {
status: self.status.as_u16(),
url: self.url.to_string(),
body: String::from_utf8_lossy(&self.body)
.chars()
.take(1_024)
.collect(),
})
}
pub fn body_lossy(&self) -> String {
String::from_utf8_lossy(&self.body).into_owned()
}
}
/// Cheaply cloneable client; Reqwest internally shares its connection pool.
#[derive(Clone)]
pub struct HttpClient {
inner: reqwest::Client,
options: HttpClientConfig,
}
impl HttpClient {
pub fn new(options: &HttpClientConfig) -> Result<Self> {
let redirect = if options.follow_redirects {
Policy::limited(10)
} else {
Policy::none()
};
let inner = reqwest::Client::builder()
.redirect(redirect)
.timeout(options.timeout())
// Disabling certificate checks is dangerous. The option is useful
// for controlled local environments; verification stays enabled.
.danger_accept_invalid_certs(!options.verify_ssl)
.build()?;
Ok(Self {
inner,
options: options.clone(),
})
}
pub async fn get(&self, url: &Url) -> Result<HttpResponse> {
self.request(Method::GET, url, HeaderMap::new(), None).await
}
pub async fn get_with_user_agent(&self, url: &Url, agent: &str) -> Result<HttpResponse> {
let mut headers = HeaderMap::new();
headers.insert(
USER_AGENT,
HeaderValue::from_str(agent).map_err(|error| {
CrawlError::Configuration(format!("invalid user-agent header: {error}"))
})?,
);
self.request(Method::GET, url, headers, None).await
}
pub async fn post_json<T: Serialize + ?Sized>(
&self,
url: &Url,
headers: &[(&str, &str)],
value: &T,
) -> Result<HttpResponse> {
let mut header_map = HeaderMap::new();
for (name, value) in headers {
let name = HeaderName::from_bytes(name.as_bytes()).map_err(|error| {
CrawlError::Configuration(format!("invalid HTTP header name: {error}"))
})?;
let value = HeaderValue::from_str(value).map_err(|error| {
CrawlError::Configuration(format!("invalid HTTP header value: {error}"))
})?;
header_map.insert(name, value);
}
self.request(
Method::POST,
url,
header_map,
Some(serde_json::to_value(value)?),
)
.await
}
async fn request(
&self,
method: Method,
url: &Url,
headers: HeaderMap,
json: Option<serde_json::Value>,
) -> Result<HttpResponse> {
let max_attempts = self.options.max_retries + 1;
for attempt in 0..max_attempts {
let mut request = self
.inner
.request(method.clone(), url.clone())
.headers(headers.clone());
if !headers.contains_key(USER_AGENT) {
request = request.header(
USER_AGENT,
user_agent::choose(self.options.rotate, &self.options.user_agent),
);
}
if let Some(value) = &json {
request = request.json(value);
}
match request.send().await {
Ok(response) => {
let status = response.status();
let response_url = response.url().clone();
let response_headers = response.headers().clone();
if is_transient(status) && attempt + 1 < max_attempts {
self.delay(attempt, Some(&response_headers)).await;
continue;
}
let body = response.bytes().await?.to_vec();
return Ok(HttpResponse {
status,
headers: response_headers,
url: response_url,
body,
});
}
Err(error) if attempt + 1 < max_attempts => {
tracing::warn!(attempt = attempt + 1, %url, %error, "HTTP transport failed; retrying");
self.delay(attempt, None).await;
}
Err(error) => return Err(error.into()),
}
}
unreachable!("the retry loop always returns on its last attempt")
}
async fn delay(&self, attempt: u32, headers: Option<&HeaderMap>) {
let retry_after = headers
.filter(|_| self.options.respect_retry_after)
.and_then(|headers| headers.get(RETRY_AFTER))
.and_then(|value| value.to_str().ok())
.and_then(parse_retry_after);
sleep(retry_after.unwrap_or_else(|| self.backoff(attempt))).await;
}
fn backoff(&self, attempt: u32) -> Duration {
let base = (self.options.backoff.initial
* self.options.backoff.multiplier.powi(attempt as i32))
.min(self.options.backoff.max);
let jitter = rand::rng().random_range(0.0..=base * 0.25);
Duration::from_secs_f64(base + jitter)
}
}
fn is_transient(status: StatusCode) -> bool {
matches!(status.as_u16(), 408 | 425 | 429 | 500 | 502 | 503 | 504)
}
fn parse_retry_after(value: &str) -> Option<Duration> {
if let Ok(seconds) = value.parse::<u64>() {
return Some(Duration::from_secs(seconds));
}
let target = httpdate::parse_http_date(value).ok()?;
Some(target.duration_since(SystemTime::now()).unwrap_or_default())
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn retry_after_accepts_seconds() {
assert_eq!(parse_retry_after("12"), Some(Duration::from_secs(12)));
}
#[test]
fn transient_statuses_are_explicit() {
assert!(is_transient(StatusCode::TOO_MANY_REQUESTS));
assert!(!is_transient(StatusCode::NOT_FOUND));
}
}
+130
View File
@@ -0,0 +1,130 @@
//! Open Graph metadata extraction.
use scraper::{Html, Selector};
use url::Url;
use crate::{
domain::ArticleMetadata,
error::Result,
http::{HttpClient, user_agent::OPEN_GRAPH_USER_AGENT},
};
/// Fetch a page with the Open Graph crawler user-agent and parse its metadata.
pub async fn consume_url(client: &HttpClient, url: &Url) -> Result<Option<ArticleMetadata>> {
let response = client
.get_with_user_agent(url, OPEN_GRAPH_USER_AGENT)
.await?
.require_success()?;
Ok(consume_html(&response.text()?, url))
}
/// Extract metadata without a network request when a caller already has HTML.
pub fn consume_html(html: &str, page_url: &Url) -> Option<ArticleMetadata> {
if html.trim().is_empty() {
return None;
}
let document = Html::parse_document(html);
let metadata = ArticleMetadata {
title: pick([meta(&document, "og:title"), text(&document, "title")]),
description: pick([
meta(&document, "og:description"),
meta(&document, "description"),
]),
image: pick([
meta(&document, "og:image"),
attribute(&document, "img", "src"),
])
.and_then(|value| absolute_url(page_url, &value)),
url: pick([
meta(&document, "og:url"),
attribute(&document, "link[rel='canonical']", "href"),
Some(page_url.as_str().to_owned()),
])
.and_then(|value| absolute_url(page_url, &value)),
author: pick([
meta(&document, "article:author"),
meta(&document, "og:article:author"),
]),
published_at: pick([
meta(&document, "article:published_time"),
meta(&document, "og:article:published_time"),
]),
updated_at: pick([
meta(&document, "article:modified_time"),
meta(&document, "og:article:modified_time"),
]),
};
(!metadata.is_empty()).then_some(metadata)
}
// Selector parsing can fail, so helpers return `None` rather than panicking on
// malformed markup or an accidentally invalid selector.
fn meta(document: &Html, property: &str) -> Option<String> {
let selector = Selector::parse(&format!(
"meta[property='{property}'], meta[name='{property}']"
))
.ok()?;
document
.select(&selector)
.next()?
.value()
.attr("content")
.map(str::to_owned)
}
fn text(document: &Html, selector: &str) -> Option<String> {
let selector = Selector::parse(selector).ok()?;
let value = document
.select(&selector)
.next()?
.text()
.collect::<Vec<_>>()
.join(" ");
(!value.trim().is_empty()).then(|| value.trim().to_owned())
}
fn attribute(document: &Html, selector: &str, name: &str) -> Option<String> {
let selector = Selector::parse(selector).ok()?;
document
.select(&selector)
.next()?
.value()
.attr(name)
.map(str::to_owned)
}
fn pick<const N: usize>(values: [Option<String>; N]) -> Option<String> {
values
.into_iter()
.flatten()
.map(|value| value.trim().to_owned())
.find(|value| !value.is_empty())
}
fn absolute_url(base: &Url, value: &str) -> Option<Url> {
Url::parse(value).or_else(|_| base.join(value)).ok()
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn extracts_open_graph_and_resolves_relative_urls() {
let html = r#"
<html><head>
<meta property="og:title" content="A title">
<meta property="og:image" content="/image.jpg">
<link rel="canonical" href="/story">
</head></html>
"#;
let base = Url::parse("https://example.com/news/page").unwrap();
let metadata = consume_html(html, &base).unwrap();
assert_eq!(metadata.title.as_deref(), Some("A title"));
assert_eq!(
metadata.image.unwrap().as_str(),
"https://example.com/image.jpg"
);
assert_eq!(metadata.url.unwrap().as_str(), "https://example.com/story");
}
}
+31
View File
@@ -0,0 +1,31 @@
//! User-agent selection.
//!
//! Rotation is not an anonymity mechanism. It only mirrors the original
//! crawler's compatibility behavior for sites that reject uncommon clients.
use rand::prelude::IndexedRandom;
pub(crate) const OPEN_GRAPH_USER_AGENT: &str =
"facebookexternalhit/1.1 (+http://www.facebook.com/externalhit_uatext.php)";
const USER_AGENTS: &[&str] = &[
"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 Chrome/131 Safari/537.36",
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 Chrome/131 Safari/537.36",
"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 Chrome/131 Safari/537.36",
"Mozilla/5.0 (iPhone; CPU iPhone OS 18_1 like Mac OS X) AppleWebKit/605.1.15 Mobile/15E148",
];
pub(crate) fn choose(rotate: bool, fallback: &str) -> String {
if !rotate {
return fallback.to_owned();
}
// `choose` returns an Option because it also supports empty slices. Our
// static list is non-empty, but keeping the fallback makes the invariant
// explicit instead of relying on `unwrap`.
USER_AGENTS
.choose(&mut rand::rng())
.copied()
.unwrap_or(fallback)
.to_owned()
}
+30
View File
@@ -0,0 +1,30 @@
//! Basango's reusable crawler library.
//!
//! Think of this file as a map, not a storage room. It declares the top-level
//! modules and keeps implementation details in focused files. Most callers
//! only need [`Crawler`] and [`CrawlRequest`].
mod articles;
mod cli;
pub mod config;
mod crawler;
pub mod domain;
pub mod error;
mod execution;
mod http;
mod sources;
mod telemetry;
pub use articles::{DeliveryStatus, Outbox, OutboxEntry, normalize};
pub use crawler::Crawler;
pub use domain::{
Article, ArticleDraft, ArticleMetadata, CrawlRequest, DateRange, PageRange, SourceId,
UpdateDirection,
};
pub use error::{CrawlError, Result};
pub use execution::CrawlReport;
/// Run the bundled command-line interface.
pub async fn run_cli() -> anyhow::Result<()> {
cli::run().await
}
+13
View File
@@ -0,0 +1,13 @@
//! The executable crate for Basango.
//!
//! A Cargo package can contain both a library crate (`lib.rs`) and a binary
//! crate (`main.rs`). Keeping this file tiny is intentional: the library owns
//! the application logic, while this binary only provides the Tokio runtime and
//! converts an error into a non-zero process exit.
#[tokio::main]
async fn main() -> anyhow::Result<()> {
// `basango` refers to the sibling library crate. We do not write
// `mod lib;`: that would incorrectly make `lib.rs` a child of this binary.
basango::run_cli().await
}
+79
View File
@@ -0,0 +1,79 @@
//! Source adapters for HTML sites and WordPress REST APIs.
//!
//! Both adapters stream the same domain value (`ArticleDraft`). A bounded
//! channel lets collection overlap persistence while applying backpressure
//! when SQLite or the backend is slower than the source website.
mod common;
mod html;
mod wordpress;
use serde::{Deserialize, Serialize};
use serde_json::Value;
use tokio::sync::mpsc;
use url::Url;
use crate::{
config::SourceConfig,
domain::{ArticleDraft, CrawlRequest},
error::Result,
http::HttpClient,
};
/// A discovery result that can be serialized into an article-fetch job.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ArticleSeed {
pub url: Url,
#[serde(skip_serializing_if = "Option::is_none")]
pub data: Option<Value>,
}
pub enum SourceAdapter {
Html(html::HtmlCrawler),
WordPress(wordpress::WordPressCrawler),
}
impl SourceAdapter {
pub fn new(source: SourceConfig, http: HttpClient) -> Self {
match source {
SourceConfig::Html(config) => Self::Html(html::HtmlCrawler::new(config, http)),
SourceConfig::WordPress(config) => {
Self::WordPress(wordpress::WordPressCrawler::new(config, http))
}
}
}
/// Start collecting immediately and return a bounded stream of results.
pub fn stream(mut self, request: CrawlRequest) -> mpsc::Receiver<Result<ArticleDraft>> {
const BUFFERED_ARTICLES: usize = 32;
let (sender, receiver) = mpsc::channel(BUFFERED_ARTICLES);
tokio::spawn(async move {
let result = match &mut self {
Self::Html(crawler) => crawler.crawl_into(&request, &sender).await,
Self::WordPress(crawler) => crawler.crawl_into(&request, &sender).await,
};
if let Err(error) = result {
let _ = sender.send(Err(error)).await;
}
});
receiver
}
pub async fn discover(&mut self, request: &CrawlRequest) -> Result<Vec<ArticleSeed>> {
match self {
Self::Html(crawler) => crawler.discover(request).await,
Self::WordPress(crawler) => crawler.discover(request).await,
}
}
pub async fn collect(
&mut self,
seed: &ArticleSeed,
request: &CrawlRequest,
) -> Result<ArticleDraft> {
match self {
Self::Html(crawler) => crawler.collect(&seed.url, request).await,
Self::WordPress(crawler) => crawler.collect(seed, request).await,
}
}
}
+57
View File
@@ -0,0 +1,57 @@
//! Parsing helpers shared by source implementations.
use chrono::{DateTime, NaiveDate, NaiveDateTime, TimeZone, Utc};
use scraper::Html;
use url::Url;
/// Resolve a link against its source while preserving already-absolute URLs.
pub(crate) fn absolute_url(base: &Url, value: &str) -> Option<Url> {
Url::parse(value).or_else(|_| base.join(value)).ok()
}
/// Convert a JavaScript/date-fns-oriented source format into common Chrono
/// formats. Unknown formats still fall back to RFC 3339 and a few safe defaults.
pub(crate) fn parse_published_at(raw: &str, configured_format: &str) -> Option<DateTime<Utc>> {
let value = raw.trim();
if value.is_empty() {
return None;
}
if let Ok(date) = DateTime::parse_from_rfc3339(value) {
return Some(date.with_timezone(&Utc));
}
if let Ok(date) = DateTime::parse_from_rfc2822(value) {
return Some(date.with_timezone(&Utc));
}
let chrono_format = match configured_format {
"dd.MM.yyyy" => "%d.%m.%Y",
"yyyy-LL-dd" => "%Y-%m-%d",
"yyyy-LL-dd HH:mm" => "%Y-%m-%d %H:%M",
"yyyy-LL-dd'T'HH:mm:ss" => "%Y-%m-%dT%H:%M:%S",
other => other,
};
if let Ok(value) = NaiveDateTime::parse_from_str(value, chrono_format) {
// A source-local timezone is not always provided. Treating naive values
// as UTC is deterministic; offset-bearing values above retain offsets.
return Some(Utc.from_utc_datetime(&value));
}
if let Ok(value) = NaiveDate::parse_from_str(value, chrono_format) {
return value
.and_hms_opt(0, 0, 0)
.map(|value| Utc.from_utc_datetime(&value));
}
// WordPress commonly omits its timezone suffix.
NaiveDateTime::parse_from_str(value, "%Y-%m-%dT%H:%M:%S")
.ok()
.map(|value| Utc.from_utc_datetime(&value))
}
pub(crate) fn text_from_html(html: &str) -> Option<String> {
let fragment = Html::parse_fragment(html);
let text = fragment.root_element().text().collect::<Vec<_>>().join(" ");
let text = text.split_whitespace().collect::<Vec<_>>().join(" ");
(!text.is_empty()).then_some(text)
}
+383
View File
@@ -0,0 +1,383 @@
//! Generic CSS-selector-driven HTML crawler.
use std::{collections::HashSet, time::Duration};
use regex::Regex;
use scraper::{ElementRef, Html, Selector};
use tokio::sync::mpsc;
use tokio::time::sleep;
use url::Url;
use crate::{
config::HtmlSourceConfig,
domain::{ArticleDraft, CrawlRequest, PageRange},
error::{CrawlError, Result},
http::{HttpClient, consume_open_graph_html},
sources::{ArticleSeed, common},
};
pub struct HtmlCrawler {
source: HtmlSourceConfig,
http: HttpClient,
}
impl HtmlCrawler {
pub fn new(source: HtmlSourceConfig, http: HttpClient) -> Self {
Self { source, http }
}
/// Crawl listings and detail pages directly in one process.
pub async fn crawl_into(
&self,
request: &CrawlRequest,
sender: &mpsc::Sender<Result<ArticleDraft>>,
) -> Result<()> {
let page_range = match request.page_range {
Some(range) => range,
None => self.pagination(request.category.as_deref()).await?,
};
for page in page_range.start..=page_range.end {
let endpoint = self.endpoint_url(page, request.category.as_deref())?;
let listing = match self.fetch_text(&endpoint).await {
Ok(listing) => listing,
Err(error) => {
tracing::error!(%error, %endpoint, page, "failed to fetch HTML listing");
continue;
}
};
let entries = self.listing_entries(&listing)?;
if entries.is_empty() {
tracing::warn!(page, %endpoint, "HTML listing contained no matching articles");
}
for entry in entries {
let Some(link) = self.extract_link(&entry)? else {
tracing::warn!(page, "skipping HTML listing entry without a link");
continue;
};
let html = if self.source.fetch_details {
match self.fetch_text(&link).await {
Ok(html) => html,
Err(error) => {
tracing::error!(%error, %link, "failed to fetch HTML detail page");
continue;
}
}
} else {
entry.html
};
match self.parse_article(&html, Some(&link), request.category.as_deref()) {
Ok(draft) => {
if let Some(range) = request.date_range {
if range.is_older_than_range(draft.published_at) {
// Listings are newest-first. This is a control
// signal, not a failure, so return collected data.
return Ok(());
}
if !range.contains(draft.published_at) {
continue;
}
}
if sender.send(Ok(draft)).await.is_err() {
return Ok(());
}
}
Err(error) => tracing::error!(%error, %link, "failed to parse HTML article"),
}
}
}
Ok(())
}
/// Discover detail URLs for the Redis-backed execution mode.
pub async fn discover(&self, request: &CrawlRequest) -> Result<Vec<ArticleSeed>> {
let page_range = match request.page_range {
Some(range) => range,
None => self.pagination(request.category.as_deref()).await?,
};
let mut locations = Vec::new();
let mut seen = HashSet::new();
for page in page_range.start..=page_range.end {
let endpoint = self.endpoint_url(page, request.category.as_deref())?;
let listing = self.fetch_text(&endpoint).await?;
for entry in self.listing_entries(&listing)? {
if let Some(url) = self.extract_link(&entry)?
&& seen.insert(url.clone())
{
locations.push(ArticleSeed { url, data: None });
}
}
}
Ok(locations)
}
pub async fn collect(&self, url: &Url, request: &CrawlRequest) -> Result<ArticleDraft> {
let html = self.fetch_text(url).await?;
let draft = self.parse_article(&html, Some(url), request.category.as_deref())?;
if request
.date_range
.is_some_and(|range| !range.contains(draft.published_at))
{
return Err(CrawlError::ArticleOutOfDateRange {
url: url.to_string(),
});
}
Ok(draft)
}
pub fn endpoint_url(&self, page: u32, category: Option<&str>) -> Result<Url> {
let mut template = self.source.pagination_template.clone();
template = template.replace("{category}", category.unwrap_or_default());
if template.contains("{page}") {
template = template.replace("{page}", &page.to_string());
}
let mut url =
common::absolute_url(&self.source.common.url, &template).ok_or_else(|| {
CrawlError::Configuration(format!("invalid pagination template '{template}'"))
})?;
if !self.source.pagination_template.contains("{page}") && page > 0 {
url.query_pairs_mut().append_pair("page", &page.to_string());
}
Ok(url)
}
async fn pagination(&self, category: Option<&str>) -> Result<PageRange> {
let fallback = PageRange::new(0, 1)?;
let url = self.endpoint_url(0, category)?;
let Ok(html) = self.fetch_text(&url).await else {
return Ok(fallback);
};
let document = Html::parse_document(&html);
let selector = parse_selector(&self.source.selectors.pagination)?;
let Some(href) = document
.select(&selector)
.filter_map(|element| element.value().attr("href"))
.next_back()
else {
return Ok(fallback);
};
let absolute = common::absolute_url(&self.source.common.url, href);
let page = absolute
.as_ref()
.and_then(|url| url.query_pairs().find(|(key, _)| key == "page"))
.and_then(|(_, value)| value.parse::<u32>().ok())
.or_else(|| {
Regex::new(r"(?:page[=/]|[?&]page=)(\d+)")
.expect("static regex is valid")
.captures(href)
.and_then(|captures| captures.get(1))
.and_then(|value| value.as_str().parse().ok())
})
.unwrap_or(1);
PageRange::new(0, page.max(1))
}
fn listing_entries(&self, html: &str) -> Result<Vec<ListingEntry>> {
let document = Html::parse_document(html);
let selector = parse_selector(&self.source.selectors.list)?;
Ok(document
.select(&selector)
.map(|element| ListingEntry {
html: element.html(),
})
.collect())
}
fn extract_link(&self, entry: &ListingEntry) -> Result<Option<Url>> {
let fragment = Html::parse_fragment(&entry.html);
let selector = parse_selector(&self.source.selectors.link)?;
let value = fragment.select(&selector).next().and_then(|element| {
element
.value()
.attr("href")
.or_else(|| element.value().attr("data-href"))
.or_else(|| element.value().attr("src"))
});
Ok(value.and_then(|value| common::absolute_url(&self.source.common.url, value)))
}
fn parse_article(
&self,
html: &str,
known_url: Option<&Url>,
selected_category: Option<&str>,
) -> Result<ArticleDraft> {
let document = Html::parse_document(html);
let title = extract_text(&document, &self.source.selectors.title)?
.ok_or_else(|| CrawlError::InvalidArticle("missing article title".into()))?;
let link = known_url
.cloned()
.or_else(|| {
extract_attribute(&document, &self.source.selectors.link)
.ok()
.flatten()
.and_then(|value| common::absolute_url(&self.source.common.url, &value))
})
.ok_or_else(|| CrawlError::InvalidArticle("missing article link".into()))?;
let raw_date = extract_text(&document, &self.source.selectors.date)?
.ok_or_else(|| CrawlError::InvalidArticle("missing article date".into()))?;
let published_at = common::parse_published_at(&raw_date, &self.source.common.date_format)
.ok_or_else(|| {
CrawlError::InvalidArticle(format!("cannot parse article date '{raw_date}'"))
})?;
let body_selector = parse_selector(&self.source.selectors.body)?;
let parts: Vec<String> = document
.select(&body_selector)
.map(|node| html2md::parse_html(&node.html()))
.filter(|part| !part.trim().is_empty())
.collect();
let body = if parts.is_empty() {
html2md::parse_html(html)
} else {
parts.join("\n")
};
let categories = self.extract_categories(&document, selected_category)?;
let metadata = consume_open_graph_html(html, &link);
Ok(ArticleDraft {
title,
body,
link,
source_id: self.source.common.id.clone(),
categories,
metadata,
published_at,
})
}
fn extract_categories(&self, document: &Html, fallback: Option<&str>) -> Result<Vec<String>> {
let Some(selector) = &self.source.selectors.categories else {
return Ok(fallback
.map(|category| vec![category.to_lowercase()])
.unwrap_or_default());
};
let selector = parse_selector(selector)?;
let mut seen = HashSet::new();
Ok(document
.select(&selector)
.filter_map(element_text)
.map(|value| value.to_lowercase())
.filter(|value| seen.insert(value.clone()))
.collect())
}
async fn fetch_text(&self, url: &Url) -> Result<String> {
if self.source.common.rate_limit {
// The original config only carries a boolean. One second is a
// conservative default until a per-source duration is introduced.
sleep(Duration::from_secs(1)).await;
}
self.http.get(url).await?.require_success()?.text()
}
}
struct ListingEntry {
html: String,
}
fn parse_selector(value: &str) -> Result<Selector> {
Selector::parse(value).map_err(|error| {
CrawlError::InvalidSourceSelectors(format!("selector '{value}' is invalid: {error}"))
})
}
fn extract_text(document: &Html, selector: &str) -> Result<Option<String>> {
let selector = parse_selector(selector)?;
Ok(document.select(&selector).next().and_then(|element| {
let name = element.value().name();
let special = match name {
"img" => element
.value()
.attr("alt")
.or_else(|| element.value().attr("title")),
"time" => element.value().attr("datetime"),
"meta" => element.value().attr("content"),
_ => None,
};
special
.map(str::trim)
.filter(|value| !value.is_empty())
.map(str::to_owned)
.or_else(|| element_text(element))
}))
}
fn extract_attribute(document: &Html, selector: &str) -> Result<Option<String>> {
let selector = parse_selector(selector)?;
Ok(document.select(&selector).next().and_then(|element| {
element
.value()
.attr("href")
.or_else(|| element.value().attr("data-href"))
.or_else(|| element.value().attr("src"))
.map(str::to_owned)
}))
}
fn element_text(element: ElementRef<'_>) -> Option<String> {
let value = element.text().collect::<Vec<_>>().join(" ");
let value = value.split_whitespace().collect::<Vec<_>>().join(" ");
(!value.is_empty()).then_some(value)
}
#[cfg(test)]
mod tests {
use crate::config::{CommonSourceConfig, HtmlSelectors};
use super::*;
fn source() -> HtmlSourceConfig {
HtmlSourceConfig {
common: CommonSourceConfig {
id: crate::domain::SourceId::new("example").unwrap(),
url: Url::parse("https://example.com").unwrap(),
..CommonSourceConfig::default()
},
fetch_details: false,
pagination_template: "news/{page}".into(),
selectors: HtmlSelectors {
body: ".body".into(),
categories: Some(".category".into()),
date: "time".into(),
link: "a".into(),
list: ".article".into(),
title: "h1".into(),
pagination: ".pages a".into(),
},
}
}
#[test]
fn parses_an_html_article_without_network_access() {
let http = HttpClient::new(&Default::default()).unwrap();
let crawler = HtmlCrawler::new(source(), http);
let html = r#"
<html><head><meta property="og:title" content="Metadata title"></head>
<body><h1>Article title</h1><time datetime="2025-02-01T12:00:00Z"></time>
<div class="body"><p>Hello <strong>world</strong></p></div>
<span class="category">Politics</span></body></html>
"#;
let url = Url::parse("https://example.com/story").unwrap();
let article = crawler.parse_article(html, Some(&url), None).unwrap();
assert_eq!(article.title, "Article title");
assert!(article.body.contains("Hello"));
assert_eq!(article.categories, vec!["politics"]);
}
#[test]
fn substitutes_category_and_page_in_endpoint() {
let mut source = source();
source.pagination_template = "category/{category}/page/{page}".into();
let crawler = HtmlCrawler::new(source, HttpClient::new(&Default::default()).unwrap());
assert_eq!(
crawler.endpoint_url(3, Some("news")).unwrap().as_str(),
"https://example.com/category/news/page/3"
);
}
}
+366
View File
@@ -0,0 +1,366 @@
//! WordPress REST API source adapter.
use std::{collections::HashMap, time::Duration};
use reqwest::header::HeaderMap;
use serde::{Deserialize, Serialize};
use tokio::sync::mpsc;
use tokio::time::sleep;
use url::Url;
use crate::{
config::{MetadataStrategy, WordPressSourceConfig},
domain::{ArticleDraft, ArticleMetadata, CrawlRequest, PageRange},
error::{CrawlError, Result},
http::{HttpClient, consume_open_graph_url},
sources::{ArticleSeed, common},
};
const POST_FIELDS: &str =
"date,slug,link,title.rendered,content.rendered,excerpt.rendered,categories,yoast_head_json";
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct RenderedField {
#[serde(default)]
pub rendered: String,
}
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct WordPressPost {
#[serde(default)]
pub categories: Vec<u64>,
#[serde(default)]
pub content: RenderedField,
pub date: Option<String>,
#[serde(default)]
pub excerpt: RenderedField,
pub link: Option<Url>,
pub slug: Option<String>,
#[serde(default)]
pub title: RenderedField,
pub yoast_head_json: Option<YoastMetadata>,
}
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct YoastMetadata {
pub article_modified_time: Option<String>,
pub article_published_time: Option<String>,
pub author: Option<String>,
pub description: Option<String>,
pub og_description: Option<String>,
#[serde(default)]
pub og_image: Vec<YoastImage>,
pub og_title: Option<String>,
pub og_url: Option<String>,
pub title: Option<String>,
}
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct YoastImage {
pub url: Option<String>,
}
pub struct WordPressCrawler {
source: WordPressSourceConfig,
http: HttpClient,
categories: HashMap<u64, String>,
}
impl WordPressCrawler {
pub fn new(source: WordPressSourceConfig, http: HttpClient) -> Self {
Self {
source,
http,
categories: HashMap::new(),
}
}
pub async fn crawl_into(
&mut self,
request: &CrawlRequest,
sender: &mpsc::Sender<Result<ArticleDraft>>,
) -> Result<()> {
let range = match request.page_range {
Some(range) => range,
None => self.pagination().await?,
};
for page in range.start..=range.end {
let posts = match self.fetch_page(page).await {
Ok(posts) => posts,
Err(error) => {
tracing::error!(%error, page, source = %self.source.common.id, "failed to fetch WordPress page");
continue;
}
};
for post in posts {
match self.post_to_draft(&post).await {
Ok(draft) => {
if let Some(range) = request.date_range {
if range.is_older_than_range(draft.published_at) {
return Ok(());
}
if !range.contains(draft.published_at) {
continue;
}
}
if sender.send(Ok(draft)).await.is_err() {
return Ok(());
}
}
Err(error) => tracing::error!(%error, "failed to parse WordPress article"),
}
}
}
Ok(())
}
pub async fn discover(&self, request: &CrawlRequest) -> Result<Vec<ArticleSeed>> {
let range = match request.page_range {
Some(range) => range,
None => self.pagination().await?,
};
let mut locations = Vec::new();
for page in range.start..=range.end {
for post in self.fetch_page(page).await? {
if let Some(url) = post.link.clone() {
locations.push(ArticleSeed {
url,
data: Some(serde_json::to_value(post)?),
});
}
}
}
Ok(locations)
}
pub async fn collect(
&mut self,
seed: &ArticleSeed,
request: &CrawlRequest,
) -> Result<ArticleDraft> {
let value = seed.data.as_ref().ok_or_else(|| {
CrawlError::InvalidArticle("WordPress details job is missing its REST payload".into())
})?;
let post: WordPressPost = serde_json::from_value(value.clone())?;
let draft = self.post_to_draft(&post).await?;
if request
.date_range
.is_some_and(|range| !range.contains(draft.published_at))
{
return Err(CrawlError::ArticleOutOfDateRange {
url: seed.url.to_string(),
});
}
Ok(draft)
}
async fn post_to_draft(&mut self, post: &WordPressPost) -> Result<ArticleDraft> {
let link = post
.link
.clone()
.ok_or_else(|| CrawlError::InvalidArticle("missing WordPress article link".into()))?;
let title = common::text_from_html(&post.title.rendered)
.or_else(|| post.slug.clone())
.unwrap_or_else(|| "Untitled".into());
let raw_date = post
.date
.as_deref()
.ok_or_else(|| CrawlError::InvalidArticle("missing WordPress article date".into()))?;
let published_at = common::parse_published_at(raw_date, &self.source.common.date_format)
.ok_or_else(|| {
CrawlError::InvalidArticle(format!("cannot parse WordPress date '{raw_date}'"))
})?;
let categories = self.map_categories(&post.categories).await;
let metadata = self.metadata(post, &link).await;
Ok(ArticleDraft {
title,
body: html2md::parse_html(&post.content.rendered),
link,
source_id: self.source.common.id.clone(),
categories,
metadata,
published_at,
})
}
async fn metadata(&self, post: &WordPressPost, link: &Url) -> Option<ArticleMetadata> {
let strategy = self.source.metadata_strategy;
let extracted = match strategy {
MetadataStrategy::None | MetadataStrategy::Fetch => None,
MetadataStrategy::Yoast => yoast_metadata(post, link),
MetadataStrategy::Rest => rest_metadata(post),
MetadataStrategy::Auto => yoast_metadata(post, link).or_else(|| rest_metadata(post)),
};
let should_fetch = matches!(strategy, MetadataStrategy::Fetch)
|| matches!(strategy, MetadataStrategy::Auto) && extracted.is_none();
if should_fetch {
consume_open_graph_url(&self.http, link)
.await
.ok()
.flatten()
} else {
extracted
}
}
async fn pagination(&self) -> Result<PageRange> {
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 pages = header_number(&response.headers, "x-wp-totalpages").unwrap_or(1);
let posts = header_number(&response.headers, "x-wp-total").unwrap_or(0);
tracing::info!(
pages,
posts,
source = %self.source.common.id,
"WordPress pagination"
);
PageRange::new(1, pages.max(1))
}
fn page_url(&self, page: u32) -> Result<Url> {
let mut url = self.api_url("wp-json/wp/v2/posts")?;
url.query_pairs_mut()
.append_pair("_fields", POST_FIELDS)
.append_pair("orderby", "date")
.append_pair("order", "desc")
.append_pair("page", &page.to_string())
.append_pair("per_page", "100");
Ok(url)
}
async fn fetch_page(&self, page: u32) -> Result<Vec<WordPressPost>> {
self.fetch(&self.page_url(page)?)
.await?
.require_success()?
.json()
}
async fn fetch_categories(&mut self) -> Result<()> {
let mut url = self.api_url("wp-json/wp/v2/categories")?;
url.query_pairs_mut()
.append_pair("_fields", "id,slug,count")
.append_pair("orderby", "count")
.append_pair("order", "desc")
.append_pair("per_page", "100");
let categories: Vec<WordPressCategory> =
self.fetch(&url).await?.require_success()?.json()?;
self.categories
.extend(categories.into_iter().map(|item| (item.id, item.slug)));
Ok(())
}
async fn map_categories(&mut self, ids: &[u64]) -> Vec<String> {
if self.categories.is_empty()
&& let Err(error) = self.fetch_categories().await
{
tracing::warn!(%error, "failed to fetch WordPress categories");
}
let mut ids = ids.to_vec();
ids.sort_unstable();
ids.into_iter()
.filter_map(|id| self.categories.get(&id).cloned())
.collect()
}
fn api_url(&self, path: &str) -> Result<Url> {
// `Url::join` treats a base without a trailing slash as a file. Force a
// directory base so a configured subpath is not accidentally replaced.
let mut base = self.source.common.url.clone();
if !base.path().ends_with('/') {
let path = format!("{}/", base.path());
base.set_path(&path);
}
base.join(path).map_err(Into::into)
}
async fn fetch(&self, url: &Url) -> Result<crate::http::HttpResponse> {
if self.source.common.rate_limit {
sleep(Duration::from_secs(1)).await;
}
self.http.get(url).await
}
}
#[derive(Debug, Deserialize)]
struct WordPressCategory {
id: u64,
slug: String,
}
fn header_number(headers: &HeaderMap, name: &str) -> Option<u32> {
headers.get(name)?.to_str().ok()?.parse().ok()
}
fn yoast_metadata(post: &WordPressPost, link: &Url) -> Option<ArticleMetadata> {
let yoast = post.yoast_head_json.as_ref()?;
let metadata = ArticleMetadata {
author: pick([yoast.author.clone()]),
description: pick([yoast.og_description.clone(), yoast.description.clone()]),
image: pick([yoast.og_image.iter().find_map(|image| image.url.clone())])
.and_then(|value| common::absolute_url(link, &value)),
published_at: pick([yoast.article_published_time.clone(), post.date.clone()]),
title: pick([yoast.og_title.clone(), yoast.title.clone()]),
updated_at: pick([yoast.article_modified_time.clone()]),
url: pick([yoast.og_url.clone(), Some(link.to_string())])
.and_then(|value| common::absolute_url(link, &value)),
};
(!metadata.is_empty()).then_some(metadata)
}
fn rest_metadata(post: &WordPressPost) -> Option<ArticleMetadata> {
let metadata = ArticleMetadata {
title: common::text_from_html(&post.title.rendered),
description: common::text_from_html(&post.excerpt.rendered),
url: post.link.clone(),
published_at: post.date.clone(),
..ArticleMetadata::default()
};
(!metadata.is_empty()).then_some(metadata)
}
fn pick<const N: usize>(values: [Option<String>; N]) -> Option<String> {
values
.into_iter()
.flatten()
.map(|value| value.trim().to_owned())
.find(|value| !value.is_empty())
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn extracts_yoast_metadata() {
let link = Url::parse("https://example.com/story").unwrap();
let post = WordPressPost {
link: Some(link.clone()),
yoast_head_json: Some(YoastMetadata {
og_title: Some("Yoast title".into()),
og_image: vec![YoastImage {
url: Some("/cover.jpg".into()),
}],
..YoastMetadata::default()
}),
..WordPressPost::default()
};
let metadata = yoast_metadata(&post, &link).unwrap();
assert_eq!(metadata.title.as_deref(), Some("Yoast title"));
assert_eq!(
metadata.image.unwrap().as_str(),
"https://example.com/cover.jpg"
);
}
#[test]
fn parses_wordpress_naive_datetime_as_utc() {
let date: chrono::DateTime<chrono::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");
}
}
+10
View File
@@ -0,0 +1,10 @@
//! Operational telemetry emitted by crawler agents.
//!
//! Signals describe what the crawler observed. The API owns the database
//! projection used by the operations dashboard.
mod reporter;
mod signal;
pub(crate) use reporter::{AgentReporter, RunReporter};
pub(crate) use signal::RunMetrics;
+190
View File
@@ -0,0 +1,190 @@
use std::env;
use chrono::Utc;
use uuid::Uuid;
use crate::{
articles::endpoint_url,
config::IngestionApiConfig,
http::HttpClient,
telemetry::signal::{IngestionSignal, RunSignalContext},
};
use super::RunMetrics;
/// Telemetry available to a long-lived worker agent.
#[derive(Clone)]
pub struct AgentReporter {
publisher: SignalPublisher,
}
impl AgentReporter {
pub fn new(config: &IngestionApiConfig, client: HttpClient) -> Self {
Self {
publisher: SignalPublisher::new(config, client),
}
}
pub async fn heartbeat(&self) {
self.publisher.heartbeat().await;
}
}
/// Telemetry scoped to one source run.
///
/// Keeping run identity in this type makes it impossible for agent-only code
/// to accidentally emit a run signal without a run or source identifier.
#[derive(Clone)]
pub struct RunReporter {
publisher: SignalPublisher,
run_id: String,
source_id: String,
}
impl RunReporter {
pub fn new(config: &IngestionApiConfig, client: HttpClient, source_id: &str) -> Self {
Self {
publisher: SignalPublisher::new(config, client),
run_id: Uuid::now_v7().to_string(),
source_id: source_id.to_owned(),
}
}
pub async fn heartbeat(&self) {
self.publisher.heartbeat().await;
}
pub async fn preparing(&self) {
self.publisher
.publish(IngestionSignal::RunPreparing {
context: self.context(),
})
.await;
}
pub async fn started(&self) {
self.publisher
.publish(IngestionSignal::RunStarted {
context: self.context(),
})
.await;
}
pub async fn progress(&self, metrics: RunMetrics) {
self.publisher
.publish(IngestionSignal::RunProgress {
context: self.context(),
metrics,
})
.await;
}
pub async fn completed(&self, metrics: RunMetrics, duration_ms: u64) {
self.publisher
.publish(IngestionSignal::RunCompleted {
context: self.context(),
metrics,
duration_ms,
})
.await;
}
pub async fn failed(&self, metrics: RunMetrics, duration_ms: u64, error: String) {
self.publisher
.publish(IngestionSignal::RunFailed {
context: self.context(),
metrics,
duration_ms,
error,
})
.await;
}
fn context(&self) -> RunSignalContext {
RunSignalContext {
signal_id: signal_id(),
agent_id: self.publisher.agent_id.clone(),
emitted_at: Utc::now(),
version: self.publisher.version.clone(),
run_id: self.run_id.clone(),
source_id: self.source_id.clone(),
}
}
}
#[derive(Clone)]
struct SignalPublisher {
agent_id: String,
client: HttpClient,
endpoint: Option<url::Url>,
token: String,
version: String,
}
impl SignalPublisher {
fn new(config: &IngestionApiConfig, client: HttpClient) -> Self {
let endpoint = config
.endpoint
.as_ref()
.and_then(|base| endpoint_url(base, "ingest/signals").ok());
Self {
agent_id: agent_id(),
client,
endpoint,
token: config.token.clone(),
version: env!("CARGO_PKG_VERSION").to_owned(),
}
}
async fn heartbeat(&self) {
self.publish(IngestionSignal::AgentHeartbeat {
signal_id: signal_id(),
agent_id: self.agent_id.clone(),
emitted_at: Utc::now(),
version: self.version.clone(),
})
.await;
}
async fn publish(&self, signal: IngestionSignal) {
match serde_json::to_string(&signal) {
Ok(serialized) => tracing::info!(signal = serialized, "ingestion signal"),
Err(error) => tracing::warn!(%error, "could not serialize ingestion signal"),
}
let Some(endpoint) = &self.endpoint else {
return;
};
let headers = [("Authorization", self.token.as_str())];
match self.client.post_json(endpoint, &headers, &signal).await {
Ok(response) if response.is_success() => {}
Ok(response) => tracing::warn!(
status = %response.status,
body = %response.body_lossy(),
"ingestion API rejected a signal"
),
Err(error) => tracing::warn!(%error, "could not publish ingestion signal"),
}
}
}
fn signal_id() -> String {
Uuid::now_v7().to_string()
}
fn agent_id() -> String {
if let Ok(value) = env::var("BASANGO_CRAWLER_AGENT_ID") {
if !value.trim().is_empty() {
return value;
}
}
// Keep accepting the former variable while deployments migrate.
if let Ok(value) = env::var("BASANGO_CRAWLER_NODE_ID") {
if !value.trim().is_empty() {
return value;
}
}
env::var("HOSTNAME")
.or_else(|_| env::var("COMPUTERNAME"))
.unwrap_or_else(|_| format!("crawler-{}", std::process::id()))
}
+103
View File
@@ -0,0 +1,103 @@
use chrono::{DateTime, Utc};
use serde::Serialize;
#[derive(Debug, Clone, Copy, Default, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct RunMetrics {
pub articles_discovered: usize,
pub articles_persisted: usize,
pub articles_delivered: usize,
pub articles_failed: usize,
}
#[derive(Debug, Clone, Serialize)]
#[serde(tag = "type")]
pub enum IngestionSignal {
#[serde(rename = "agent.heartbeat")]
AgentHeartbeat {
#[serde(rename = "signalId")]
signal_id: String,
#[serde(rename = "agentId")]
agent_id: String,
#[serde(rename = "emittedAt")]
emitted_at: DateTime<Utc>,
version: String,
},
#[serde(rename = "run.preparing")]
RunPreparing {
#[serde(flatten)]
context: RunSignalContext,
},
#[serde(rename = "run.started")]
RunStarted {
#[serde(flatten)]
context: RunSignalContext,
},
#[serde(rename = "run.progress")]
RunProgress {
#[serde(flatten)]
context: RunSignalContext,
metrics: RunMetrics,
},
#[serde(rename = "run.completed")]
RunCompleted {
#[serde(flatten)]
context: RunSignalContext,
metrics: RunMetrics,
#[serde(rename = "durationMs")]
duration_ms: u64,
},
#[serde(rename = "run.failed")]
RunFailed {
#[serde(flatten)]
context: RunSignalContext,
metrics: RunMetrics,
#[serde(rename = "durationMs")]
duration_ms: u64,
error: String,
},
}
#[derive(Debug, Clone, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct RunSignalContext {
pub signal_id: String,
pub agent_id: String,
pub emitted_at: DateTime<Utc>,
pub version: String,
pub run_id: String,
pub source_id: String,
}
#[cfg(test)]
mod tests {
use chrono::TimeZone;
use super::*;
#[test]
fn serializes_a_discriminated_progress_signal() {
let signal = IngestionSignal::RunProgress {
context: RunSignalContext {
signal_id: "signal-1".into(),
agent_id: "agent-1".into(),
emitted_at: Utc.with_ymd_and_hms(2026, 8, 23, 12, 0, 0).unwrap(),
version: "1.0.0".into(),
run_id: "run-1".into(),
source_id: "source-1".into(),
},
metrics: RunMetrics {
articles_discovered: 3,
articles_persisted: 2,
articles_delivered: 1,
articles_failed: 0,
},
};
let value = serde_json::to_value(signal).unwrap();
assert_eq!(value["type"], "run.progress");
assert_eq!(value["signalId"], "signal-1");
assert_eq!(value["metrics"]["articlesDelivered"], 1);
assert!(value.get("event").is_none());
}
}
+37
View File
@@ -0,0 +1,37 @@
//! Integration test for the durable article pipeline.
//!
//! Source-specific HTML/WordPress parsing has focused unit tests. This test
//! crosses public module boundaries: draft → normalization → outbox persistence.
use basango::{ArticleDraft, DeliveryStatus, Outbox, SourceId, normalize};
use chrono::Utc;
use tempfile::tempdir;
use url::Url;
#[test]
fn draft_is_normalized_and_persisted_for_later_delivery() {
let directory = tempdir().unwrap();
let sqlite_path = directory.path().join("crawler.db");
let draft = ArticleDraft {
title: " Fixture\u{00a0}story ".into(),
body: "Hello from Rust.\n\n\nSecond paragraph.".into(),
link: Url::parse("https://example.com/story").unwrap(),
source_id: SourceId::new("fixture").unwrap(),
categories: vec!["Learning".into(), "learning".into()],
metadata: None,
published_at: Utc::now(),
};
// `None` means no ingestion API client is configured. The outbox therefore
// becomes the durable hand-off point for a later `push` command.
let article = normalize(draft).unwrap();
let outbox = Outbox::open(&sqlite_path, true).unwrap();
let status = outbox.save(&article).unwrap();
assert_eq!(status, DeliveryStatus::Pending);
assert_eq!(article.title, "Fixture story");
assert_eq!(article.categories, vec!["Learning"]);
let pending = outbox.list_pending(Some("fixture"), 10).unwrap();
assert_eq!(pending.len(), 1);
assert_eq!(pending[0].article, article);
}