# CLAUDE.md This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository. ## Project Overview **go-caatsm** is a Civil Aviation Authority Telegram Message Processor - a high-performance, production-ready message processing system for aviation telegrams. It uses Clean Architecture, NATS JetStream for reliable message streaming, and PostgreSQL/TimescaleDB for persistence. The system processes ICAO-format aviation telegrams (ARR, DEP, CNL, DLA, FPL), parsing raw messages from NATS JetStream, storing them in PostgreSQL, and republishing structured JSON to downstream consumers. ## Essential Commands ### Build & Run ```bash # Build binary (outputs to bin/receiver) make build # or: task build VERSION=v1.0.0 make build # with custom version # Development (JetStream mode, auto-creates stream/consumer) make run-dev # or: task run-dev make run-local # or: task run-local (go run, honors GO_ENV) # Production (requires config.prod.toml, manual stream/consumer setup) make run-prod # or: task run-prod ``` ### Testing ```bash # Unit tests (Ginkgo, verbose) make test # or: task test ginkgo -r -v ./cmd ./internal # Run single test by description ginkgo -r -v --focus "Test Description" ./path/to/package # Integration tests (requires Docker) make test-int # or: task test-int # All tests make test-all # or: task test-all # Coverage report (target >80%) make coverage # or: task coverage # Opens coverage/coverage.html ``` ### Code Quality ```bash # Lint (golangci-lint) make lint # or: task lint # Format code make fmt # or: task fmt go fmt ./... # Regenerate dependency injection (after modifying pkg/di/wire.go) make wire # or: task wire wire ./pkg/di ``` ### Development Tools ```bash # Install Ginkgo tooling task install-test go install github.com/onsi/ginkgo/v2/ginkgo@latest # Generate sample telegrams (burst mode) make seed COUNT=100 CATEGORY=mixed STATUS=random # or: task seed # Continuous slow seeding (useful for monitoring) make seed-slow INTERVAL_MIN=2s INTERVAL_MAX=5s # or: task seed-slow # Press Ctrl-C to stop # Start dev infrastructure (Docker Compose) task up # Postgres, NATS, observability stack task dev-run # Start infra + run receiver locally task down # Stop and remove containers ``` ## Architecture Overview ### Clean Architecture Layers ``` internal/ ├── domain/ # Pure business entities (no dependencies) │ └── aviation.go # ARR, DEP, CNL, DLA, FPL, Weather domain models ├── port/ # Interface contracts (Repository, Publisher) │ ├── repository.go │ └── publisher.go ├── app/ # Application logic (orchestration) │ └── processor.go # MessageProcessor - main processing pipeline ├── adapter/ # Interface implementations & data transformations │ ├── parser/ # Telegram parsers (composite pattern) │ │ ├── aviation/ # Aviation telegrams (ARR, DEP, CNL, DLA, FPL) │ │ ├── weather/ # Weather reports (METAR, SPECI, TAF) │ │ └── schedule/ # Flight schedule messages │ ├── mapper/ # Domain ↔ DTO transformations │ └── dto/ # Data Transfer Objects (ParsedTelegram, MessageStatus) └── infra/ # Infrastructure concerns ├── nats/ # NATS JetStream consumer & publisher ├── postgres/ # PostgreSQL repository (pgx, batch inserts) ├── config/ # Koanf configuration loading ├── log/ # Zap structured logging ├── telemetry/ # OpenTelemetry traces/metrics ├── metrics/ # Prometheus metrics ├── monitoring/ # Health/metrics HTTP server └── buildinfo/ # Build metadata ``` **Dependency Flow**: `infra` → `adapter` → `app` → `domain` ← `port` The domain layer has zero external dependencies. All layers depend on interfaces (ports), not concrete implementations. ### Key Components 1. **NATS Consumer** (`internal/infra/nats/consumer.go`) - Pulls messages from JetStream in batches (default: 50) - ACK/NAK handling with automatic retries - Routes to DLQ after max delivery attempts (default: 3) - Monitors consumer lag and pending messages 2. **MessageProcessor** (`internal/app/processor.go`) - Orchestrates: Parse → Persist → Publish - Error categorization: `parsed`, `header_error`, `body_error`, `publish_error`, `repository_error` - Parser failures are permanent (ACK'd, no retry) - Repository failures are transient (NAK'd, retry with backoff) 3. **Parser** (`internal/adapter/parser/`) - Composite parser architecture with specialized sub-parsers - **Aviation Parser** (`internal/adapter/parser/aviation/`) - ICAO telegram formats (ARR, DEP, CNL, DLA, FPL) - **Weather Parser** (`internal/adapter/parser/weather/`) - Weather reports (METAR, SPECI, TAF) - **Schedule Parser** (`internal/adapter/parser/schedule/`) - Flight schedule messages - Uses pattern matching, tokenization, and lexical analysis - Returns structured domain models or error status 4. **Repository** (`internal/infra/postgres/repository.go`) - PostgreSQL persistence via pgx connection pool - Batch insert support via `COPY FROM` - Stores parsed telegrams in `aviation.telegrams` table 5. **Monitoring Server** (`internal/infra/monitoring/server.go`) - `GET /livez` - Liveness (process info, no dependencies checked) - `GET /readyz` - Readiness (pings Postgres & NATS, 503 on failure) - `GET /metrics` - Prometheus metrics endpoint - Default address: `:2112` ### Configuration Uses Koanf for config loading from TOML files + environment variables: - Config file: `configs/config.{GO_ENV}.toml` (GO_ENV defaults to `dev`) - Env overrides: `CAATSM_` prefix (e.g., `CAATSM_NATS_URL`, `CAATSM_POSTGRES_URL`) - Key settings: - `nats.mode`: Must be `"jetstream"` (only supported mode) - `nats.stream`: JetStream stream name (default: `TELEGRAM`) - `nats.consumer`: Durable consumer name (default: `telegram-consumer`) - `nats.consumer_rules.max_deliver`: Max retry attempts (default: 3) - `dlq.enabled`: Enable Dead-Letter Queue for poison messages - `app.batch_size`: JetStream pull batch size (default: 50) - `monitoring.addr`: Metrics/health server address (default: `:2112`) ### Parser Architecture The system uses a **composite parser pattern** with specialized sub-parsers: 1. **Composite Parser** (`internal/adapter/parser/composite.go`) - Orchestrates multiple specialized parsers - Routes messages to appropriate parser based on content classification - Falls back gracefully if primary parser fails 2. **Aviation Parser** (`internal/adapter/parser/aviation/`) - Pattern-based parsing using registry of message type patterns - Tokenizer for breaking down telegram structure - Handles ARR, DEP, CNL, DLA, FPL message types - Extracts flight details, aircraft info, timestamps, airports 3. **Weather Parser** (`internal/adapter/parser/weather/`) - Lexer-based parsing for weather reports - Classifier to identify METAR, SPECI, or TAF format - Pattern matching for weather elements (wind, visibility, clouds, etc.) - Normalizes weather data into structured format 4. **Schedule Parser** (`internal/adapter/parser/schedule/`) - Extracts flight schedule information - Pattern matching for schedule-specific fields - Handles recurring flight patterns and time ranges Each parser implements the `Parser` interface from `internal/port/parser.go`, enabling easy extension and testing. ### Dependency Injection Uses Google Wire for compile-time DI: - Wire spec: `pkg/di/wire.go` - Generated code: `pkg/di/wire_gen.go` (NEVER edit manually) - To add a new dependency: 1. Create a `ProvideXxx` function in the appropriate package 2. Add it to the `runtimeSet` in `pkg/di/wire.go` 3. Run `make wire` or `task wire` ### Message Processing Flow ``` NATS JetStream → Consumer.Fetch(batch) → MessageProcessor.Handle(msg) → ├─ Parser.Parse(raw) → domain model or error status ├─ Repository.Insert(parsed) → Postgres aviation.telegrams ├─ Publisher.Publish(json) → NATS output topic └─ ACK (success) | NAK (transient failure) | ACK (permanent failure) ``` **Error Handling**: - **Parser failures** (invalid format): Store raw + status in DB, ACK message (permanent) - **Repository failures** (DB down): NAK message, JetStream redelivers (transient) - **Publisher failures** (downstream topic): Store raw + error status, ACK message (permanent, can replay from DB) - After `max_deliver` attempts: Route to DLQ if enabled, else discard ### Observability Three observability surfaces: 1. **Prometheus Metrics** (`/metrics`) - `caatsm_messages_total{stream,consumer,result}` - Throughput & results - `caatsm_handle_latency_seconds_bucket{stream,consumer}` - Processing latency - `caatsm_retries_total{stream,consumer,reason}` - JetStream retries - `caatsm_db_queries_total{operation,result}` - DB activity - `caatsm_nats_consumer_pending_messages{stream,consumer}` - Consumer lag - `caatsm_dlq_messages_total{stream,consumer}` - DLQ routing 2. **OpenTelemetry** (traces + metrics) - Environment-based sampling: Production (1%), Staging (10%), Dev/Test (100%) - Semantic attributes: `messaging.system`, `db.system`, `caatsm.message.category` - Configure via `[telemetry]` block (endpoint, enabled, insecure) - Export to OTLP/HTTP collector (default: `localhost:4318`) 3. **Health Endpoints** - `GET /livez` - Process liveness (no dependency checks) - `GET /readyz` - Readiness (checks Postgres & NATS, timeout: 2s) - Both return JSON with build metadata + dependency status ### Testing Philosophy - **Unit tests**: Ginkgo BDD style, table-driven, mock interfaces - Location: Alongside code (`*_test.go`) - Run with `make test` or `ginkgo -r -v ./cmd ./internal` - Target: >80% coverage - **Integration tests**: Docker-based (testcontainers), full E2E flow - Location: `test/integration/` - Run with `make test-int` (requires Docker) - Spins up real Postgres + NATS JetStream - **Benchmarks**: Performance-critical paths (parser, mapper, processor) - Run with `go test -bench=BenchmarkXxx -benchmem ./path/to/pkg` ## Important Development Notes ### Code Style (from .cursor/rules/do.mdc) - **Clean Architecture**: Strict layer separation, dependency inversion - **Interface-driven**: All public functions interact with interfaces, not concrete types - **Error handling**: Always wrap errors with context (`fmt.Errorf("context: %w", err)`) - **Context propagation**: Pass `context.Context` everywhere (deadlines, cancellations, tracing) - **No global state**: Use constructor functions with DI - **Resource cleanup**: Use `defer` for closing resources - **Security**: Input validation, secure defaults, retries with exponential backoff - **Observability**: Trace all service boundaries (HTTP, NATS, DB), structured logs (JSON) ### Never Edit Generated Files - `pkg/di/wire_gen.go` - Wire-generated DI code - Any `*_gen.go` files - Code generation output ### Common Gotchas 1. **JetStream-only mode**: The system only supports JetStream mode (`nats.mode = "jetstream"`). Do not attempt to use Core NATS mode. 2. **Production setup**: In production (`GO_ENV=prod`): - Stream and Consumer must be created manually (not auto-created) - Requires `configs/config.prod.toml` or `CAATSM_*` env vars - Enable TLS for NATS/Postgres connections 3. **Parsing failures are permanent**: Parser errors (invalid format) are ACK'd and stored with error status. They do NOT trigger JetStream retries. 4. **Repository failures are transient**: DB connection failures trigger NAK, causing JetStream to redeliver the message. 5. **Batch processing**: Default batch size is 50 messages. Tune `app.batch_size` for your workload. Larger batches improve throughput but increase latency. 6. **Consumer lag**: Monitor `caatsm_nats_consumer_pending_messages` metric. High values indicate slow processing or insufficient consumer instances. 7. **DLQ routing**: Enable DLQ (`dlq.enabled = true`) to capture poison messages after max delivery attempts. Inspect DLQ subject (`caatsm.dlq`) for failed messages. ## Additional Documentation - `README.md` - Comprehensive user guide (setup, configuration, deployment) - `AGENTS.md` - Quick reference for AI assistants (commands, style guide) - `docs/prod-guide.md` - Production deployment guide - `docs/nats.md` - NATS/JetStream configuration details - `docs/observability.md` - Observability setup & metrics - `docs/migrations.md` - Database migration strategy - `docs/performance.md` - Performance tuning guidelines