News Pipeline Engine

An always-on news collection, extraction, publication, and physical alerting service. It collects RSS, search, and Telegram inputs; validates and scrapes public web pages; asks language models to extract structured stories; stores durable records through Prism; publishes Redis views; and delivers important alerts to Discord, a siren, and a Kindle.

The application is designed for a single trusted operator on a private network. It runs on Bun, is scheduled with BullMQ, and relies on Redis plus several local gRPC services.

Non-negotiable behavior

These are the system's core invariants:

Every newly accepted alert synchronously fans out to Discord, siren, and Kindle.
Alert delivery has no category, geography, genre, severity, source-count, or trusted-domain restriction.
A failing alert channel does not prevent the other two channels from being attempted.
A URL becomes visited only after the resulting data is durably saved or durably staged.
Scraping accepts only credential-free public HTTP(S) destinations.
Prism/SQL is the durable story store. Redis news collections are rebuildable publication views.
Queue startup never destroys the complete BullMQ database.

System architecture

flowchart LR
    subgraph Inputs
        RSS[Google News RSS]
        SEARCH[SearchService]
        TG[Telegram OSINT in Redis]
    end

    subgraph Runtime["Bun process"]
        QP[(news-pipeline)]
        QH[(news-heartbeat)]
        PRIMARY[Primary pipeline]
        SECONDARY[Secondary pipeline]
        POLICY[Public URL policy]
        SCRAPE[Isolated browser scraping]
        CLEAN[ScraperService cleanup]
        LLM[Structured extraction agents]
        SAVE[Save-news tool]
        ALERT[Alert tool]
        CLEANUP[Nightly cleanup]
    end

    subgraph Storage
        REDIS[(Redis)]
        PRISM[(Prism / SQL)]
    end

    subgraph Outputs
        DISCORD[Discord]
        SIREN[Siren]
        KINDLE[Kindle]
        KUMA[Uptime Kuma]
    end

    RSS --> PRIMARY
    SEARCH --> SECONDARY
    TG --> PRIMARY
    QP --> PRIMARY
    QP --> SECONDARY
    QP --> CLEANUP
    QH --> KUMA
    PRIMARY --> POLICY
    SECONDARY --> POLICY
    POLICY --> SCRAPE --> CLEAN
    CLEAN --> LLM
    SECONDARY --> REDIS
    REDIS --> PRIMARY
    LLM --> SAVE
    LLM --> ALERT
    SAVE --> PRISM
    SAVE --> REDIS
    ALERT --> DISCORD
    ALERT --> SIREN
    ALERT --> KINDLE
    CLEANUP --> REDIS

Service boundaries

Component	Default address	Responsibility
Redis	`redis://127.0.0.1:6379`	BullMQ, staging, visited URLs, publication views, delivery checkpoints
DiscordWebhook	`127.0.0.1:50051`	Send alert text to Discord
PrismService	`127.0.0.1:50052`	Search and durably ingest stories
SearchService	`127.0.0.1:50053`	Supply secondary discovery URLs
ScraperService	`127.0.0.1:50057`	Clean/summarize scraped article content
Kindle SSH	`root@192.168.0.10`	Render alert text on the device
Siren HTTP service	private-network URL	Trigger the audible alert
Uptime Kuma	configured push URLs	Process and job liveness monitoring

All active gRPC calls have deadlines. Read-only gRPC calls get one bounded retry; writes do not retry blindly.

Pipeline behavior

Primary pipeline

sequenceDiagram
    autonumber
    participant R as RSS feeds
    participant U as URL policy
    participant B as Browser workers
    participant S as ScraperService
    participant X as Redis staging
    participant P as Prism
    participant A as Primary agent
    participant T as Save/alert tools

    R->>U: Resolved article URLs
    U->>U: Scheme, address, DNS, redirect checks
    U->>B: Public HTTP(S) URLs only
    par Isolated bounded workers
        B->>S: Scraped article text
        S-->>B: Clean content
    end
    X->>A: Pending secondary batches
    X->>A: Claimed Telegram items
    B->>A: Clean primary articles
    A->>P: Search existing coverage
    P-->>A: Related stories
    A->>T: Structured story/tool calls
    T->>P: Ingest story with idempotency key
    P-->>T: Durable success
    T->>X: Atomically publish Redis views
    T->>X: Mark successful URLs visited
    A-->>X: Acknowledge staged inputs after success

The primary run:

Loads configurable Google News RSS feeds.
Resolves article links and applies the public URL policy.
Scrapes with bounded concurrency and a fresh browser context per worker.
Cleans content through ScraperService.
Claims recoverable Telegram items and reads durable secondary batches.
Looks for prior Prism coverage.
Runs the primary agent with run-scoped source mappings.
Writes to Prism before atomically publishing Redis views.
Marks URLs visited and acknowledges pending inputs only after success.

Empty runs stop before invoking the model.

Secondary pipeline

flowchart TD
    TERMS[Configured search terms] --> SEARCH[SearchService]
    SEARCH --> FILTER{Visited or manually<br/>blacklisted?}
    FILTER -->|Yes| SKIP[Skip]
    FILTER -->|No| POLICY{Public URL?}
    POLICY -->|No| REJECT[Reject]
    POLICY -->|Yes| SCRAPE[Scrape and clean]
    SCRAPE --> HISTORY[Compare with Prism history]
    HISTORY --> EXTRACT[One structured result per article]
    EXTRACT --> QUALIFY{Qualifies?}
    QUALIFY -->|No| DONE[Discard]
    QUALIFY -->|Yes| STAGE[(secondary_news_batches)]
    STAGE --> VISITED[(visited_url)]
    STAGE --> PRIMARY[Next primary run]

Secondary extraction uses one schema-validated response with exactly one result for each input article. Qualifying items are appended to a durable Redis list. The URL is marked visited only after that append succeeds. The primary pipeline acknowledges a batch only after it finishes successfully.

Alert delivery

Alert fan-out is unconditional for every newly accepted alert. The model/tool schema validates the alert payload, but there are no business eligibility gates between acceptance and the three output channels.

flowchart TD
    START[Validated alert tool call] --> ID[Compute deterministic alert ID]
    ID --> FANOUT{{Start all incomplete channels}}
    FANOUT --> D[Discord]
    FANOUT --> S[Siren]
    FANOUT --> K[Kindle]
    D --> DS{Settled}
    S --> SS{Settled}
    K --> KS{Settled}
    DS --> JOIN[Wait for all three]
    SS --> JOIN
    KS --> JOIN
    JOIN --> RESULT{Any failure?}
    RESULT -->|No| OK[Record alert once and return success]
    RESULT -->|Yes| ERR[Return aggregate failure]
    ERR --> RETRY[Identical retry]
    RETRY --> FANOUT

Per-channel completion is checkpointed under alert_delivery:<id> for seven days. On an identical retry, already completed channels are not repeated, while failed channels are attempted again. This prevents a Kindle retry from retriggering a siren that already succeeded. The initial delivery still starts all three channels independently and waits for all three to settle.

Kindle delivery additionally provides:

fixed executable invocation rather than arbitrary shell command execution;
POSIX-safe argument quoting;
SSH connection and command deadlines;
pinned SSH host-key fingerprints;
stale connection cleanup;
a token-owned, expiring Redis display lock;
lock refresh while a display operation is active;
owner-checked Lua release.

Queues and schedules

flowchart LR
    CLOCK[Wall clock<br/>NEWS_TIMEZONE] --> P[(news-pipeline)]
    CLOCK --> H[(news-heartbeat)]
    P --> PRI[Primary worker task]
    P --> SEC[Secondary worker task]
    P --> CLN[Cleanup worker task]
    H --> HB[Heartbeat worker task]
    PRI --> JOBPING[News-job monitor]
    SEC --> JOBPING
    HB --> PROCESSPING[Process heartbeat monitor]

Job	Cron pattern	Queue	Meaning
Primary news	`5,35 7-22 * * *`	`news-pipeline`	At minute 5 and 35, 07:00–22:59
Secondary news	`45 7-22 * * *`	`news-pipeline`	At minute 45, 07:00–22:59
Cleanup	`0 0 * * *`	`news-pipeline`	Midnight
Heartbeat	`* * * * *`	`news-heartbeat`	Every minute

The timezone defaults to Asia/Kolkata. Startup also enqueues one immediate primary run.

Pipeline work and heartbeat work use separate workers so a long scrape cannot starve the minute heartbeat. Startup removes only known scheduler IDs from the legacy news queue; it does not obliterate queue state. By default, each queue retains its latest 100 completed and 100 failed jobs.

Data model and lifecycle

flowchart LR
    INPUT[Unprocessed input] --> STAGED[Durably staged]
    STAGED --> SAVED[Prism / SQL saved]
    SAVED --> PUBLISHED[Redis published views]
    PUBLISHED --> EXPIRE[Nightly view cleanup]
    SAVED -. durable history remains .-> FUTURE[Future deduplication/search]
    FAILED[Run failure] --> STAGED
    FAILED --> RETRY[Recoverable retry input]
    RETRY --> SAVED

Redis key	Kind	Lifecycle and purpose
`news_collection_v2`	list	Global published feed; cleared nightly
`news_collection_v2:<genre>`	list	Per-genre published feed; cleared nightly
`genre`	set	Published genre index; cleared nightly
`alerts_send`	collection	Alert history view; cleared nightly
`secondary_news_batches`	list	Durable unacknowledged secondary work; preserved
`telegram_retry`	list	Telegram claims restored after failure; preserved
`visited_url`	set	Successfully saved or durably staged URLs; preserved
`blacklisted_domains`	set	Explicit/manual domain exclusions; preserved
`published_story:<id>`	string	Seven-day Redis publication idempotency marker
`alert_delivery:<id>`	hash	Seven-day per-channel alert checkpoints
`kindle:lock`	string	Expiring token-owned Kindle display lease

Nightly cleanup derives every genre-specific key from the current genre set and uses Redis UNLINK to remove published views without blocking Redis on large deallocations. It preserves operational state, staging, queues, locks, visited URLs, and idempotency checkpoints.

Legacy secondary_news payloads are schema-validated and migrated into secondary_news_batches. Invalid legacy payloads are quarantined instead of being silently consumed.

Failure and recovery model

stateDiagram-v2
    [*] --> Pending
    Pending --> Processing: claim/start
    Processing --> Durable: Prism ingest or durable staging succeeds
    Processing --> Recoverable: run fails before acknowledgement
    Recoverable --> Processing: next safe attempt
    Durable --> Published: atomic Redis publication
    Published --> [*]
    Durable --> [*]: publication already checkpointed

Failure	What still happens	Recovery behavior
One alert channel fails	The other two are still attempted and awaited	Aggregate failure; identical retry attempts only incomplete channels
Prism write fails	Redis publication does not run	Job fails without falsely publishing or marking the URL visited
Redis publication repeats	Lua checks the deterministic publication marker	Duplicate Redis feed insertion is suppressed for seven days
Primary run fails after claiming Telegram	Claimed payload is restored	A later run can process `telegram_retry`
Primary run fails with secondary batches	Batch remains unacknowledged	A later primary run reads it again
Secondary scrape/save fails	URL is not marked visited	A later discovery may retry it
Read-only gRPC call times out	Deadline stops the hung call	One bounded retry is made
Process receives SIGINT/SIGTERM	Workers and queues stop accepting work	gRPC clients and Redis are closed cleanly

Automatic BullMQ retries for the full primary/secondary jobs remain disabled until Prism enforces server-side idempotent ingestion. This avoids turning an uncertain remote write into duplicate durable stories.

Security model

Web scraping boundary

Only http: and https: URLs are accepted.
Embedded URL credentials are rejected.
Direct and DNS-resolved private, loopback, link-local, multicast, reserved, unspecified, and metadata-service destinations are rejected.
Redirects and browser subrequests are checked, not only the initial URL.
Chromium error pages are rejected.
Browser sandboxing remains enabled.
Each worker uses an isolated browser context.
Scraped content is bounded before it is sent downstream.

Model boundary

Article and Telegram content is explicitly delimited and labeled as untrusted data in agent prompts. Tool inputs are schema-validated, sources must map to the current run, and unknown source identifiers fail closed.

The system still intentionally permits models to initiate automatic alerts. Prompt injection risk can be reduced but cannot be eliminated while autonomous model-triggered physical and network side effects are retained. Run the service only with trusted infrastructure and review model/provider security changes before upgrades.

Secret handling

Keep API keys and the Kindle private key in the untracked .env file or an equivalent secret store. Do not commit them, print them in logs, or include them in model input. SSH fingerprints are public verification material and may be committed.

Configuration

Bun automatically loads .env. Copy the template and fill in the required values:

cp .env.example .env

Required settings

Variable	Purpose
`OPENAI_API_KEY`	Primary model/tool-calling provider
`GOOGLE_API_KEY`	Secondary structured extraction provider
`UPTIME_KUMA_PUSH_URL`	Minute process-heartbeat push URL
`UPTIME_KUMA_NEWS_JOB_PUSH_URL`	Successful pipeline-job push URL

Runtime settings

Variable	Default	Notes
`REDIS_URL`	`redis://127.0.0.1:6379`	Canonical Redis setting for the app and BullMQ
`REDIS_HOST`, `REDIS_PORT`	localhost, `6379`	Legacy fallback only when `REDIS_URL` is absent
`NEWS_TIMEZONE`	`Asia/Kolkata`	IANA timezone used by all schedulers
`QUEUE_COMPLETED_RETENTION`	`100`	Number of completed jobs retained per queue
`QUEUE_FAILED_RETENTION`	`100`	Number of failed jobs retained per queue
`QUEUE_JOB_TIMEOUT_MS`	`1200000`	Maximum runtime for an isolated pipeline job
`GRPC_DEADLINE_MS`	`30000`	Deadline for gRPC requests
`SCRAPE_CONCURRENCY`	`4`	Maximum parallel browser workers
`SCRAPE_MAX_CONTENT_LENGTH`	`4000`	Maximum cleaned characters per scraped article
`RSS_MAX_URLS_PER_SOURCE`	`3`	Candidate cap for each RSS source
`RSS_SOURCES`	built-in Google News feeds	Pipe-separated list of feed URLs
`SECONDARY_SEARCH_TERMS`	India News, World News	Pipe-separated search queries

gRPC addresses

Variable	Default
`DISCORD_GRPC_ADDR`	`127.0.0.1:50051`
`PRISM_GRPC_ADDR`	`127.0.0.1:50052`
`SEARCH_GRPC_ADDR`	`127.0.0.1:50053`
`SCRAPER_GRPC_ADDR`	`127.0.0.1:50057`

Alert device settings

Variable	Default / format
`KINDLE_HOST`	`192.168.0.10`
`KINDLE_USER`	`root`
`KINDLE_KEY_PATH`	Absolute path to the SSH private key
`KINDLE_HOST_FINGERPRINTS`	Comma-separated `SHA256:...` fingerprints
`KINDLE_CONNECT_TIMEOUT_MS`	`10000`
`KINDLE_COMMAND_TIMEOUT_MS`	`10000`
`SIREN_URL`	`http://192.168.0.50/siren`
`ALERT_ENDPOINT_URL`	`http://192.168.0.99:8000/alert`

The repository defaults contain the fingerprints scanned from the current default Kindle. Replace them after a legitimate device or host-key change; do not disable host-key verification.

Install and run

Requirements:

Bun
Redis
Chromium compatible with the configured browser integration
the Prism, Search, Scraper, and Discord gRPC services
network access to configured alert devices and monitoring endpoints

Install dependencies:

bun install

Start the scheduler and workers:

bun run queue

The queue process validates required configuration, registers/upserts its schedulers, and submits an immediate primary job.

Run a pipeline directly for development:

bun src/data/pipeline.ts
bun src/data/secondaryPipeline.ts

Operations

launchd

com.news.queue.plist is the macOS service definition. It contains machine-specific paths, so inspect them before installation. It uses a 30-second throttle to avoid a tight crash/restart loop.

cp com.news.queue.plist ~/Library/LaunchAgents/com.news.queue.plist
launchctl bootstrap gui/$(id -u) ~/Library/LaunchAgents/com.news.queue.plist

After changing code or environment values, restart the service through the normal launchd workflow used on the host.

Logs and rotation

Default launchd logs:

/tmp/com.news.queue.out.log
/tmp/com.news.queue.err.log

The independent heartbeat queue truncates both files in place every night at midnight in NEWS_TIMEZONE. No archives are retained. In-place truncation is intentional because launchd keeps the output file descriptors open for the lifetime of the service.

Operational checks

redis-cli PING
nc -vz 127.0.0.1 50051
nc -vz 127.0.0.1 50052
nc -vz 127.0.0.1 50053
nc -vz 127.0.0.1 50057

Inspect BullMQ failures and both launchd logs when a Uptime Kuma monitor stops receiving pushes. A healthy minute heartbeat does not prove that a news job succeeded; the two monitors intentionally represent different signals.

Testing

Run the complete local verification suite:

bun test
bun run typecheck
bun run audit

The focused tests cover:

public/private URL and DNS policy decisions;
secondary extraction ordering and sentinel filtering;
cleanup key selection;
all-channel alert fan-out and partial retry behavior;
Kindle command-injection resistance and argument quoting;
scraped-content cleaning.

Before committing, also check whitespace and patch integrity:

git diff --check

Project layout

src/
├── ai/
│   ├── agents/                 Primary and secondary model orchestration
│   ├── tools/                  Run-scoped save and alert tools
│   └── alertDelivery.ts        Unconditional three-channel delivery
├── data/
│   ├── newsCollection/         Published Redis view management
│   ├── rss/                    Feed discovery
│   ├── scrape/                 Browser extraction
│   ├── secondaryNews/          Durable secondary staging
│   ├── telegram/               Recoverable Telegram claims
│   ├── pipeline.ts             Primary pipeline
│   └── secondaryPipeline.ts    Secondary pipeline
├── db/redis/                   Shared Bun Redis client
├── grpc/                       Deadline-aware service clients and protos
├── integrations/
│   ├── kindle/                 Pinned SSH display integration
│   └── siren/                  HTTP siren integration
├── security/urlPolicy.ts       SSRF and public-address policy
├── config.ts                   Central environment configuration
└── newsQueueRoute.ts           BullMQ schedulers, workers, shutdown

Known deployment boundary

The client and prism.proto now send a deterministic IngestStoryRequest.idempotency_key. The independently deployed Prism server must persist that field and implement an upsert/unique constraint before full BullMQ pipeline retries can be enabled safely. Redis publication is already idempotent, but client-side suppression cannot prove whether a timed-out remote write committed.

License

This project is free software licensed under the GNU Affero General Public License, version 3 or any later version (AGPL-3.0-or-later). See LICENSE for the complete terms.

The AGPL is a strong copyleft license. If you modify this program and let users interact with the modified version over a network, you must offer those users the Corresponding Source as required by section 13. Preserve copyright and license notices when redistributing the software. This README is an operational summary, not a substitute for the license text.

Name		Name	Last commit message	Last commit date
Latest commit History 199 Commits
src		src
.codex		.codex
.env.example		.env.example
.gitignore		.gitignore
AGENTS.md		AGENTS.md
CLAUDE.md		CLAUDE.md
LICENSE		LICENSE
README.md		README.md
bun.lock		bun.lock
com.news.queue.plist		com.news.queue.plist
package.json		package.json
tsconfig.json		tsconfig.json

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Repository files navigation

News Pipeline Engine

Contents

Non-negotiable behavior

System architecture

Service boundaries

Pipeline behavior

Primary pipeline

Secondary pipeline

Alert delivery

Queues and schedules

Data model and lifecycle

Failure and recovery model

Security model

Web scraping boundary

Model boundary

Secret handling

Configuration

Required settings

Runtime settings

gRPC addresses

Alert device settings

Install and run

Operations

launchd

Logs and rotation

Operational checks

Testing

Project layout

Known deployment boundary

License

About

Uh oh!

Contributors

Uh oh!

Languages

Folders and files

Latest commit

History

Repository files navigation

News Pipeline Engine

Contents

Non-negotiable behavior

System architecture

Service boundaries

Pipeline behavior

Primary pipeline

Secondary pipeline

Alert delivery

Queues and schedules

Data model and lifecycle

Failure and recovery model

Security model

Web scraping boundary

Model boundary

Secret handling

Configuration

Required settings

Runtime settings

gRPC addresses

Alert device settings

Install and run

Operations

launchd

Logs and rotation

Operational checks

Testing

Project layout

Known deployment boundary

License

About

Topics

Resources

License

Uh oh!

Stars

Watchers

Forks

Contributors

Uh oh!

Languages