A newsletter/RSS aggregator with pub/sub architecture for learning and daily use.
Agregado is a personal project to learn distributed systems patterns while building a useful tool. It aggregates content from RSS feeds and email newsletters, scores articles with AI at ingest time, and delivers a ranked daily digest email with a feedback loop that improves future scoring over time.
| Component | Technology | Why |
|---|---|---|
| Backend | Go | Performance, simplicity, excellent concurrency |
| Message Broker | RabbitMQ | Classic AMQP patterns, enterprise-relevant |
| Database | PostgreSQL | Full-featured, scalable, industry standard |
| Frontend | HTMX + Go templates | Minimal JS, stays in Go ecosystem |
| Email Intake | Webhooks (Cloudflare Email Routing) | Event-driven, aligns with pub/sub goals |
| AI Inference | Cloudflare Workers AI | Scoring, summarization, digest overview |
| Deployment | Docker Compose | Self-hosted, reproducible |
╔══════════════════════════════════════════════════════════════════╗
║ INGESTION ║
╠══════════════════════════════════════════════════════════════════╣
║ ║
║ ┌─────────────┐ ┌──────────────────────────────────┐ ║
║ │ RSS Poller │ │ Email Webhook Handler │ ║
║ │ (periodic) │ │ (Cloudflare Email Routing) │ ║
║ └──────┬──────┘ └────────────┬─────────────────────┘ ║
║ │ │ ║
║ │ ┌─────────▼──────────┐ ║
║ │ │ Newsletter Article │ ║
║ │ │ (always created) │ ║
║ │ └────┬──────────┬─────┘ ║
║ │ │ │ ║
║ │ [summarize=true] [extract_links=true] ║
║ │ ┌────▼───┐ ┌───▼───────────┐ ║
║ │ │AI Sum. │ │ Link Extractor │ ║
║ │ │→summary│ │ + Fetcher │ ║
║ │ └────────┘ └───────┬────────┘ ║
║ │ │ ║
║ │ child articles ║
║ │ │ ║
╠═════════╪════════════════════════════════════════╪══════════════╣
║ │ AI SCORING │ ║
║ │ ┌──────────────────┐ │ ║
║ └─────────►│ Score(1–5) │◄─────────┘ ║
║ │ • quality │ ║
║ │ • topic weights │◄── feedback loop ║
║ └────────┬─────────┘ ║
║ │ relevance_score stored on article ║
╠═════════════════════════════╪════════════════════════════════════╣
║ QUEUE │ ║
║ ┌────────▼────────┐ ║
║ │ RabbitMQ │ ║
║ │ articles.ingest│ ║
║ └────────┬────────┘ ║
║ │ ║
║ ┌────────▼────────┐ ║
║ │ Storage Worker │ ║
║ │ (deduplicates) │ ║
║ └────────┬────────┘ ║
║ ▼ ║
║ ┌──────────┐ ║
║ │PostgreSQL│ ║
║ └──────────┘ ║
╠══════════════════════════════════════════════════════════════════╣
║ DIGEST (daily CRON) ║
╠══════════════════════════════════════════════════════════════════╣
║ ║
║ Ranker: filter score ≥ 3, sort score DESC, cap N articles ║
║ │ ║
║ ▼ ║
║ Generator: ║
║ 1. Group articles by tag ║
║ 2. Per group → AI Summarize → group summary ║
║ 3. All group summaries → AI Digest → overview paragraph ║
║ 4. Per article → HMAC token (for feedback links) ║
║ │ ║
║ ▼ ║
║ Email (SMTP): ║
║ [Overview paragraph] ║
║ ── Tag: Tech ────────────────────────────────────── ║
║ AI summary of Tech articles ║
║ • Article title 👍 👎 ║
║ ── Tag: Business ────────────────────────────────── ║
║ ... ║
╠══════════════════════════════════════════════════════════════════╣
║ FEEDBACK LOOP ║
╠══════════════════════════════════════════════════════════════════╣
║ ║
║ 👍 / 👎 links in email ║
║ │ GET /api/feedback?article_id=…&vote=up&token=… ║
║ ▼ ║
║ Validate HMAC token ║
║ │ ║
║ ├── Insert → article_feedback ║
║ └── Upsert → topic_weights (±0.1, clamped 0.1–2.0) ║
║ ▲ ║
║ └── used in next Score() prompt ║
╚══════════════════════════════════════════════════════════════════╝
RSS feeds: Poller fetches feeds on a configurable interval → parses entries → scores each article (1–5) using AI with topic weights → publishes to RabbitMQ → Storage Worker deduplicates by URL → saves to PostgreSQL.
Email newsletters: Cloudflare Email Routing → Webhook Handler → always creates a newsletter article. Two per-source toggles control additional processing:
summarize = true→ AI generates a summary of the newsletter body → stored inarticles.summaryextract_links = true→ goquery extracts links from HTML → go-readability fetches and parses each → child articles created withparent_article_idpointing to the newsletter → each child is scored and published to the queue
CRON triggers the digest pipeline:
- Rank — query articles from last 24h where
relevance_score >= 3(or unscored), sorted by score DESC, capped at N - Group — articles grouped by tag (Tech, Business, etc.)
- Summarize — AI generates a 2–3 sentence summary per tag group
- Overview — AI generates a 2-sentence intro from all group summaries ("summaries of summaries")
- Tokens — HMAC-SHA256 token generated per article per vote direction for feedback links
- Render — HTML template + plain text fallback
- Send — SMTP delivery
Each digest article includes 👍/👎 links. Clicking one:
- Validates the HMAC token (prevents tampering)
- Records a row in
article_feedback - Fetches the article's tags
- Upserts
topic_weights(up → weight += 0.1, down → weight -= 0.1, clamped 0.1–2.0)
Next time the AI scorer runs, the topic weights are passed in the prompt, biasing the score toward topics the user has engaged with.
- Go 1.22+
- Docker & Docker Compose
- Make (optional)
# Start dependencies (PostgreSQL + RabbitMQ)
make dev-deps
# Run database migrations
make migrate-up
# Start the application
make devdocker-compose upAgregado is a telemetry producer: every diagnostic is a structured record emitted as JSON on stdout. Where those logs end up is a collector's problem — the app ships no agent and stores no logs itself.
| Variable | Default | Values |
|---|---|---|
LOG_LEVEL |
info |
debug, info, warn, error — anything unrecognized falls back to info |
LOG_FORMAT |
json |
json for the deploy, text for readable local development |
# readable lines while developing
LOG_FORMAT=text LOG_LEVEL=debug make dev
# what the deploy emits
{"time":"...","level":"INFO","msg":"agregado starting","component":"main","http_port":"8080"}Every record carries a component field (main, broker, deadletter,
storage, enrich, ai, digest, api, backup, poller) — deliberately
low-cardinality, so it can become a log-store label. High-cardinality
identifiers like article_id and url are line fields, never labels.
Failures surface at ERROR the moment they happen: a handler error is logged
with its queue and a truncated body before the message is dead-lettered, and a
consumer drains articles.dlq — logging each dead message and acking it, with
no retry. Startup logs a short list of non-sensitive fields; the Config struct
is never serialized, so the database password and Cloudflare API token stay out
of the log stream.
See F17 in the PRD for the design decisions behind this.
agregado/
├── cmd/agregado/ # Application entry point
├── internal/
│ ├── ai/ # AI provider interface + Cloudflare implementation
│ ├── api/ # HTTP handlers (sources, articles, feedback, digest)
│ ├── broker/ # RabbitMQ integration (publisher, consumer)
│ ├── config/ # Configuration (env vars → structs)
│ ├── digest/ # Ranker, generator, mailer, scheduler
│ ├── domain/ # Business entities (Article, Source, Tag)
│ ├── ingestion/
│ │ ├── email/ # Webhook handler, email parser
│ │ ├── fetch/ # Link extractor + article fetcher
│ │ └── rss/ # Feed parser, poller
│ ├── logging/ # slog setup — the single logging seam
│ └── storage/ # PostgreSQL repositories
├── migrations/ # Database migrations (golang-migrate)
├── templates/ # HTMX + Go HTML templates
├── docker/ # Dockerfile
└── docs/ # PRD, ADRs, roadmap archive, study log
- Product Requirements (PRD) — Full feature specifications
- Issues — Live roadmap; what's actually being worked on
- ADRs — Architectural decisions, including deliberate non-goals
- Roadmap Archive — Frozen history of Phases 1–18
- Pub/Sub patterns — Message queues, exchanges, consumers, dead-letter queues
- Go concurrency — Goroutines, channels, graceful shutdown
- Database design — Migrations, indexes, full-text search, upsert patterns
- API design — RESTful patterns, HTMX hypermedia, HMAC token validation
- AI integration — Provider interfaces, prompt engineering, feedback loops
- DevOps basics — Docker, health checks, Cloudflare Workers
MIT