Architecture

🦉 Owl · Go Temporal PostgreSQL Kafka

Overview

Owl is a Go service (module github.com/holmes89/owl) that provides a personal digital library for books and academic papers. Three runtime processes share the same Go module:

Process	Entry point	Role
API server	`cmd/api/main.go`	Serves Connect-RPC on `:9000`; enqueues ingestion requests
Worker	`cmd/worker/main.go`	Runs ingestion pipelines + Kafka consumers
CLI	`main.go` → `cmd/`	Cobra CLI client speaking gRPC to `localhost:9000`

Process Architecture

┌────────────────────────────────────────────────┐
│  Client (CLI / curl / UI)                      │
└────────────────┬───────────────────────────────┘
                 │ Connect-RPC HTTP/2 h2c :9000
┌────────────────▼───────────────────────────────┐
│  API Server (cmd/api/main.go)                  │
│                                                │
│  BookService gRPC handler                      │
│  PaperService gRPC handler                     │
│    └─ IngestPaperByURL → Ingester.Enqueue()    │
│    └─ GetIngestionStatus → Ingester.Status()   │
│                                                │
│  Ingester (buffered channel, goroutine)        │
│  PostgreSQL (lib/repo)                         │
│  Kafka client [producer: nil]                  │
└────────────────────────────────────────────────┘
                 │ Kafka topics: owl.v1.Paper, owl.v1.Book
┌────────────────▼───────────────────────────────┐
│  Worker (cmd/worker/main.go)                   │
│                                                │
│  Kafka consumers: paper, book                  │
│    └─ message.source_url != "" → Enqueue       │
│    └─ otherwise → direct Create                │
│                                                │
│  Ingester (buffered channel, goroutine)        │
│    runPaperPipeline (7 steps)                  │
│  BookIngester (buffered channel, goroutine)    │
│    runBookPipeline (4 steps)                   │
│                                                │
│  PostgreSQL (lib/repo)                         │
│  MinIO storage (lib/storage)                   │
│  Metadata clients (lib/owl/metadata)           │
│    CompositeClient → ArxivClient               │
│                    → CrossRefClient            │
└────────────────────────────────────────────────┘

Ingestion Pipeline Detail

Paper Pipeline (`runPaperPipeline` — 7 steps)

Step	Function	Notes
1	`ResolveSourceActivity`	arXiv/DOI detection; optional MetadataClient fetch
2	`DownloadFileActivity`	HTTP GET → temp file; SHA-256 checksum
3	`ExtractMetadataActivity`	stub (returns `en`); non-fatal
4	`DeduplicateActivity`	FindByChecksum + FindByDOI; short-circuits if duplicate
5	`UploadToStorageActivity`	uploads to `papers/{checksum}`; no-op if Storage == nil
6	`PersistMetadataActivity`	PaperRepo.Create — required
7	`SubmitToMagpieActivity`	stub — non-fatal

Book Pipeline (`runBookPipeline` — 4 steps)

Step	Function	Notes
1	`DownloadFileActivity`	HTTP GET → temp file; SHA-256 checksum
2	`DeduplicateBookActivity`	FindByChecksum only
3	`UploadBookToStorageActivity`	uploads to `books/{checksum}`
4	`PersistBookActivity`	BookRepo.Create — required

Request ID Strategy

Deterministic IDs prevent duplicate concurrent ingestion and correlate API calls to status queries:

URL type	Request ID
`arxiv.org/abs/{id}`	`owl-ingest-arxiv-{id}`
`doi.org/...`	`owl-ingest-doi-{sha256[:8]}`
Other	`owl-ingest-url-{sha256[:8]}`

Ingester.Enqueue marks the request as running then Start updates to completed/failed/duplicate after the pipeline. Status is in-memory (ephemeral across restarts).

Dependency injection

cmd/api/main.go
  repo.NewDatabase(conn)
    → repo.BookRepo → book.NewBookService → bookgrpc.NewBookService(svc, nil)
    → repo.PaperRepo → paper.NewPaperService → papergrpc.NewPaperService(svc, nil)
                                                 └─ .WithIngester(ingester)
  ingestion.Activities{PaperRepo, BookRepo, MetadataClient}
    → ingestion.NewIngester(acts, 0) → go ingester.Start(ctx)

cmd/worker/main.go
  repo.NewDatabase(conn)
    → repo.BookRepo → book.NewBookService → book.NewBookConsumer(kafka, svc, bookIngester)
    → repo.PaperRepo → paper.NewPaperService → paper.NewPaperConsumer(kafka, svc, ingester)
    → ingestion.Activities{PaperRepo, BookRepo, MetadataClient, Storage}
    → ingestion.NewIngester(acts, 0)     → go ingester.Start(ctx)
    → ingestion.NewBookIngester(acts, 0) → go bookIngester.Start(ctx)

Concurrency and Error Handling

Enqueue blocks when the channel is full — the Kafka consumer goroutine provides natural backpressure; no messages are silently dropped.
Steps 3 (ExtractMetadata) and 7 (SubmitToMagpie) are non-fatal; errors are logged and the pipeline continues.
Step 6 (PersistMetadata) is required; failure aborts the pipeline for that paper.
Storage: If MINIO_ENDPOINT is unset, upload activities return a stub path; pipeline still completes.
Deduplication: If duplicate found (by checksum or DOI), pipeline returns Duplicate: true without storing again.
Status tracking (paper only): Ingester maintains an in-memory jobs map; status is lost on process restart.

Infrastructure (docker-compose)

Service	Port	Purpose
`db` (postgres:15-alpine)	internal	Owl PostgreSQL database
`minio`	`9010` (API), `9011` (console)	Object storage
`redpanda-0`	`19092`	Kafka-compatible broker
`api`	`9000`	API server
`worker`	—	Ingestion pipelines + Kafka consumers

Environment Variables

API Server (`cmd/api`)

Variable	Description
`DATABASE_URL`	PostgreSQL connection string
`KAFKA_BROKERS`	Kafka broker address

Worker (`cmd/worker`)

Variable	Description
`DATABASE_URL`	PostgreSQL connection string
`KAFKA_BROKERS`	Kafka broker address
`MINIO_ENDPOINT`	MinIO/S3 endpoint; archiving skipped if unset
`MINIO_ACCESS_KEY`	MinIO access key
`MINIO_SECRET_KEY`	MinIO secret key
`MINIO_BUCKET`	Bucket name (default: `owl`)