From 06fc9cb9e0fdf0c563c06112a7ad003cbf1c9ef4 Mon Sep 17 00:00:00 2001 From: windyboy Date: Tue, 18 Nov 2025 14:15:58 +0800 Subject: [PATCH] =?UTF-8?q?=F0=9F=94=A7=20Update=20Go=20version=20in=20go.?= =?UTF-8?q?mod=20and=20enhance=20build=20process=20with=20versioning=20inf?= =?UTF-8?q?ormation.=20Modify=20Makefile=20and=20Taskfile=20to=20inject=20?= =?UTF-8?q?build=20metadata=20(version,=20commit,=20build=20time)=20into?= =?UTF-8?q?=20the=20binary.=20Improve=20README=20with=20instructions=20for?= =?UTF-8?q?=20custom=20version=20builds=20and=20document=20new=20build=20i?= =?UTF-8?q?nfo=20features.=20Add=20benchmarks=20for=20message=20parsing=20?= =?UTF-8?q?and=20processing=20to=20improve=20performance=20testing=20capab?= =?UTF-8?q?ilities.?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- Makefile | 13 +- README.md | 54 +- Taskfile.yml | 23 +- cmd/main/main.go | 17 +- cmd/seed-telegrams/main.go | 13 +- configs/config.dev.toml | 20 + configs/config.prod.toml | 26 + docs/migrations.md | 299 ++++++++ docs/nats.md | 699 ++++++++++++++++++ docs/performance.md | 396 ++++++++++ docs/prod-guide.md | 140 ++++ docs/secret-management.md | 427 +++++++++++ go.mod | 2 +- .../parser/aviation_parser_bench_test.go | 137 ++++ internal/app/processor_bench_test.go | 147 ++++ internal/infra/config/config.go | 26 +- internal/infra/log/logger_test.go | 7 +- internal/infra/nats/advisory_dlq.go | 4 +- internal/infra/nats/consumer.go | 274 ++++++- internal/infra/nats/consumer_error.go | 6 + internal/infra/nats/consumer_jetstream.go | 20 +- internal/infra/nats/consumer_message.go | 2 + internal/infra/nats/jetstream.go | 91 ++- internal/infra/nats/utils_test.go | 45 +- internal/infra/postgres/repository.go | 6 +- .../infra/postgres/repository_bench_test.go | 169 +++++ pkg/di/wire.go | 1 - 27 files changed, 3009 insertions(+), 55 deletions(-) create mode 100644 docs/migrations.md create mode 100644 docs/performance.md create mode 100644 docs/secret-management.md create mode 100644 internal/adapter/parser/aviation_parser_bench_test.go create mode 100644 internal/app/processor_bench_test.go create mode 100644 internal/infra/postgres/repository_bench_test.go diff --git a/Makefile b/Makefile index 4a8fd46..26173b4 100644 --- a/Makefile +++ b/Makefile @@ -3,6 +3,14 @@ BUILD_DIR ?= bin BINARY := $(BUILD_DIR)/receiver CMD := ./cmd/main GO_ENV ?= dev +VERSION ?= dev + +# Build info variables +GIT_COMMIT := $(shell git rev-parse --short HEAD 2>/dev/null || echo unknown) +BUILD_TIME := $(shell printf 'package main\nimport ("fmt"\n"time")\nfunc main() { fmt.Print(time.Now().UTC().Format(time.RFC3339)) }' | go run -) +LDFLAGS := -X 'caatsm/internal/infra/buildinfo.Version=$(VERSION)' \ + -X 'caatsm/internal/infra/buildinfo.Commit=$(GIT_COMMIT)' \ + -X 'caatsm/internal/infra/buildinfo.BuiltAt=$(BUILD_TIME)' # Default target .PHONY: all @@ -12,7 +20,10 @@ all: build ## Build the application build: ## Build the receiver binary @mkdir -p $(BUILD_DIR) @echo "Building receiver..." - @go build -o $(BINARY) $(CMD) + @echo " Version: $(VERSION)" + @echo " Commit: $(GIT_COMMIT)" + @echo " Built: $(BUILD_TIME)" + @go build -ldflags "$(LDFLAGS)" -o $(BINARY) $(CMD) .PHONY: run run: run-dev ## Alias for run-dev diff --git a/README.md b/README.md index 0d4ad58..2562515 100644 --- a/README.md +++ b/README.md @@ -374,24 +374,45 @@ CAATSM_NATS_MODE=jetstream go run ./cmd/main listen ### Build +The build process automatically injects build information (version, commit, build time) into the binary. This information is available via the `/livez` and `/readyz` health endpoints. + Using Make (writes `bin/receiver`): ```bash make build +# Or with custom version: +VERSION=v1.0.0 make build ``` Using Task: ```bash task build +# Or with custom version: +VERSION=v1.0.0 task build ``` Or directly with Go: ```bash -go build -o bin/receiver ./cmd/main +# With build info injection: +go build -ldflags "-X 'caatsm/internal/infra/buildinfo.Version=dev' -X 'caatsm/internal/infra/buildinfo.Commit=$(git rev-parse --short HEAD)' -X 'caatsm/internal/infra/buildinfo.BuiltAt=$$(go run - <<'EOF' +package main +import ( + \"fmt\" + \"time\" +) +func main() { + fmt.Print(time.Now().UTC().Format(time.RFC3339)) +} +EOF)'" -o bin/receiver ./cmd/main ``` +Build information is automatically populated from: +- **Version**: `VERSION` environment variable (defaults to "dev") +- **Commit**: Git commit hash (short format) +- **BuiltAt**: UTC timestamp of build time + ### Run #### Development Mode @@ -418,8 +439,8 @@ For production deployment, see the comprehensive guide: **[Production Deployment Quick start: ```bash -# Build the binary -make build +# Build the binary with version information +VERSION=v1.0.0 make build # Run in production mode make run-prod # GO_ENV=prod (requires config.prod.toml) @@ -430,8 +451,13 @@ make run-prod # GO_ENV=prod (requires config.prod.toml) - Stream and Consumer must be created manually - Production configuration file: `configs/config.prod.toml` - SSL/TLS for secure connections +- NATS authentication configured (see `docs/prod-guide.md#nats-authentication`) -See `docs/prod-guide.md` for complete production deployment instructions. +See `docs/prod-guide.md` for complete production deployment instructions, including: +- NATS authentication setup +- Database migrations (see `docs/migrations.md`) +- Secret management (see `docs/secret-management.md`) +- Performance tuning (see `docs/performance.md`) ### Command Line Options @@ -570,6 +596,15 @@ Additional deployment-specific guides: - **`docs/deploy-systemd.md`** - Systemd service deployment with environment file configuration - **`docs/deploy-k8s.md`** - Kubernetes deployment with ConfigMap/Secret and health probes +## Additional Documentation + +- **`docs/migrations.md`** - Database migration strategy and best practices +- **`docs/secret-management.md`** - Secret management best practices and integration guides +- **`docs/performance.md`** - Performance tuning guidelines and optimization strategies +- **`docs/observability.md`** - Observability setup and metrics documentation +- **`docs/nats.md`** - NATS/JetStream configuration and usage guide +- **`docs/reliability.md`** - Reliability patterns and error handling + ### Project Structure - **Domain Layer** (`internal/domain`): Pure business logic and domain models @@ -601,6 +636,17 @@ The project keeps tests close to the code that they exercise: - **Domain/adapter/app unit tests** live under `internal/**` and cover parsing, validation, orchestration, and adapters. Run them all with `task test` (or `make test`), which now uses the Ginkgo CLI to run unit test suites in verbose mode (`ginkgo -r -v ./cmd ./internal`). - **Integration tests** under `test/integration` spin up disposable TimescaleDB and NATS JetStream instances (via `testcontainers-go`) and execute a full ingestion flow. Use `task test-int` after ensuring Docker is running. - **Coverage goals** are tracked via `task coverage`, which produces both a coverage profile and an HTML report under `coverage/coverage.html`. +- **Benchmark tests** are available for performance-critical components: + ```bash + # Run parser benchmarks + go test -bench=BenchmarkParse -benchmem ./internal/adapter/parser + + # Run repository benchmarks (requires DB connection) + go test -bench=BenchmarkMapper -benchmem ./internal/infra/postgres + + # Run processor benchmarks + go test -bench=BenchmarkHandle -benchmem ./internal/app + ``` | Purpose | Make command | Task command | |----------------------------|---------------------|---------------------| diff --git a/Taskfile.yml b/Taskfile.yml index 4d3f0b4..b704353 100644 --- a/Taskfile.yml +++ b/Taskfile.yml @@ -4,6 +4,18 @@ vars: build_dir: bin binary: '{{.build_dir}}/receiver' cmd: ./cmd/main + version: + sh: echo "${VERSION:-dev}" + git_commit: + sh: git rev-parse --short HEAD 2>/dev/null || echo unknown + build_time: + sh: | + mkdir -p .tmp + echo 'package main' > .tmp/build_time.go + echo 'import ("fmt"; "time")' >> .tmp/build_time.go + echo 'func main() { fmt.Print(time.Now().UTC().Format("2006-01-02T15:04:05Z")) }' >> .tmp/build_time.go + go run .tmp/build_time.go + rm -rf .tmp tasks: all: @@ -16,7 +28,16 @@ tasks: cmds: - mkdir -p {{.build_dir}} - echo "Building receiver..." - - go build -o {{.binary}} {{.cmd}} + - | + echo " Version: {{.version}}" + echo " Commit: {{.git_commit}}" + echo " Built: {{.build_time}}" + - | + go build \ + -ldflags "-X 'caatsm/internal/infra/buildinfo.Version={{.version}}' \ + -X 'caatsm/internal/infra/buildinfo.Commit={{.git_commit}}' \ + -X 'caatsm/internal/infra/buildinfo.BuiltAt={{.build_time}}'" \ + -o {{.binary}} {{.cmd}} run: desc: Alias for run-dev diff --git a/cmd/main/main.go b/cmd/main/main.go index bfa2491..c1035ca 100644 --- a/cmd/main/main.go +++ b/cmd/main/main.go @@ -1,6 +1,7 @@ package main import ( + "caatsm/internal/infra/buildinfo" "caatsm/internal/infra/config" "caatsm/pkg/di" "context" @@ -158,14 +159,18 @@ func runListen(parentCtx context.Context, cfg *config.Config) error { ctx, stop := signal.NotifyContext(parentCtx, os.Interrupt, syscall.SIGTERM) defer stop() - shutdownTelemetry := func(context.Context) error { return nil } + var shutdownTelemetry func(context.Context) error if cfg.Telemetry.Enabled { var telErr error shutdownTelemetry, telErr = initTelemetry(ctx, cfg) if telErr != nil { return fmt.Errorf("failed to initialize telemetry: %w", telErr) } - defer shutdownTelemetry(context.Background()) + defer func() { + if err := shutdownTelemetry(context.Background()); err != nil { + zap.L().Error("Failed to shutdown telemetry", zap.Error(err)) + } + }() } // Initialize dependencies using Wire @@ -178,7 +183,11 @@ func runListen(parentCtx context.Context, cfg *config.Config) error { if err := monitorServer.Start(ctx); err != nil { return fmt.Errorf("failed to start monitoring server: %w", err) } - defer monitorServer.Shutdown(context.Background()) + defer func() { + if err := monitorServer.Shutdown(context.Background()); err != nil { + zap.L().Error("Failed to shutdown monitoring server", zap.Error(err)) + } + }() } // Start consumer in a goroutine @@ -385,7 +394,7 @@ func initTelemetry(ctx context.Context, cfg *config.Config) (func(context.Contex resource.WithContainer(), resource.WithAttributes( semconv.ServiceName("caatsm"), - semconv.ServiceVersion("dev"), // TODO: Use build info + semconv.ServiceVersion(buildinfo.Version), semconv.ServiceNamespace("airport"), attribute.String("service.component", "receiver"), attribute.String("deployment.environment", env), diff --git a/cmd/seed-telegrams/main.go b/cmd/seed-telegrams/main.go index a6bdbb4..8aeff9b 100644 --- a/cmd/seed-telegrams/main.go +++ b/cmd/seed-telegrams/main.go @@ -62,7 +62,7 @@ func main() { duration := flag.Duration("duration", 0, "Total duration for interval/mixed modes (0 = rely on --count only)") flag.Parse() - rand.Seed(time.Now().UnixNano()) + // rand.Seed deprecated in Go 1.20+, using default source var nc *nats.Conn var js nats.JetStreamContext @@ -73,7 +73,11 @@ func main() { if err != nil { log.Fatalf("connect nats: %v", err) } - defer nc.Drain() + defer func() { + if err := nc.Drain(); err != nil { + log.Printf("failed to drain NATS connection: %v", err) + } + }() if *useJetStream { opts := []nats.JSOpt{} @@ -191,10 +195,7 @@ func runIntervalMode(cfg SeedConfig, categories []string, statuses []string, pub start := time.Now() sent := 0 - for { - if cfg.Count > 0 && sent >= cfg.Count { - break - } + for cfg.Count <= 0 || sent < cfg.Count { if cfg.Duration > 0 && time.Since(start) >= cfg.Duration { break } diff --git a/configs/config.dev.toml b/configs/config.dev.toml index c709867..a102b96 100644 --- a/configs/config.dev.toml +++ b/configs/config.dev.toml @@ -63,6 +63,26 @@ start_sequence = 0 # Example: "2024-11-15T08:00:00Z" start_time = "" +[nats.auth] +# NATS authentication configuration (optional for development) +# Only one authentication method can be used at a time: +# - token: Simple token authentication +# - credentials_file: Path to NATS credentials file (e.g., /path/to/user.creds) +# - user/password: Username and password authentication +# +# For development, authentication is typically not required. +# Uncomment and configure as needed: +# token = "" +# credentials_file = "" +# user = "" +# password = "" +# +# TLS configuration (optional) +# tls_enabled = false +# tls_cert_file = "" # Client certificate file path +# tls_key_file = "" # Client private key file path +# tls_ca_file = "" # CA certificate file for server verification + [subscription] topic = "telegram.serial" queue_group = "tele-queue" diff --git a/configs/config.prod.toml b/configs/config.prod.toml index 4f91a25..8554f02 100644 --- a/configs/config.prod.toml +++ b/configs/config.prod.toml @@ -58,6 +58,32 @@ start_sequence = 0 # Example: "2024-11-15T08:00:00Z" start_time = "" +[nats.auth] +# NATS authentication configuration (REQUIRED for production) +# Only one authentication method can be used at a time: +# - token: Simple token authentication (suitable for service-to-service) +# - credentials_file: Path to NATS credentials file (recommended for production) +# - user/password: Username and password authentication +# +# Production examples: +# +# Option 1: Token authentication +# token = "your-nats-token-here" +# +# Option 2: Credentials file (recommended) +# credentials_file = "/etc/caatsm/nats.creds" +# +# Option 3: User/Password +# user = "caatsm-service" +# password = "secure-password-here" +# +# TLS configuration (REQUIRED for production) +# Enable TLS for encrypted communication +tls_enabled = true +# tls_cert_file = "/etc/caatsm/tls/client.crt" # Client certificate file path +# tls_key_file = "/etc/caatsm/tls/client.key" # Client private key file path +# tls_ca_file = "/etc/caatsm/tls/ca.crt" # CA certificate file for server verification + [subscription] topic = "telegram.serial" queue_group = "tele-queue" diff --git a/docs/migrations.md b/docs/migrations.md new file mode 100644 index 0000000..663aebd --- /dev/null +++ b/docs/migrations.md @@ -0,0 +1,299 @@ +# Database Migrations Guide + +This document describes the database schema management and migration strategy for the CAATSM application. + +## Current Approach + +The application currently uses DDL (Data Definition Language) files for schema management: + +- **Schema file**: `internal/infra/postgres/telegrams.ddl` +- **Manual execution**: Schema changes are applied manually using `psql` or similar tools +- **Version control**: DDL files are version-controlled in the repository + +### Current Schema Structure + +The application uses TimescaleDB (PostgreSQL extension) with the following key components: + +- **Schema**: `aviation` +- **Main table**: `aviation.telegrams` (hypertable for time-series data) +- **Raw table**: `aviation.telegrams_raw` (for unparsed/failed messages) +- **Indexes**: Multiple indexes on key fields for query performance + +## Recommended Migration Tools + +For production deployments, we recommend using a dedicated migration tool for better schema management: + +### Option 1: golang-migrate (Recommended) + +[golang-migrate](https://github.com/golang-migrate/migrate) is a popular Go-based migration tool with excellent PostgreSQL support. + +**Installation:** +```bash +go install -tags 'postgres' github.com/golang-migrate/migrate/v4/cmd/migrate@latest +``` + +**Setup:** +1. Create migrations directory: + ```bash + mkdir -p migrations + ``` + +2. Create initial migration from existing schema: + ```bash + migrate create -ext sql -dir migrations -seq initial_schema + ``` + +3. Copy DDL content to migration files: + - `migrations/000001_initial_schema.up.sql` - Create schema + - `migrations/000001_initial_schema.down.sql` - Drop schema + +**Usage:** +```bash +# Apply all migrations +migrate -path migrations -database "postgres://user:pass@localhost:5432/aviation?sslmode=disable" up + +# Rollback last migration +migrate -path migrations -database "postgres://user:pass@localhost:5432/aviation?sslmode=disable" down 1 + +# Check migration version +migrate -path migrations -database "postgres://user:pass@localhost:5432/aviation?sslmode=disable" version +``` + +### Option 2: migrate (by golang-migrate, different package) + +Similar to golang-migrate but distributed as a separate package. + +### Option 3: Custom Migration Scripts + +For simple deployments, you can create custom migration scripts that: +- Check current schema version +- Apply migrations sequentially +- Track migration state in a `schema_migrations` table + +## Migration Workflow + +### Development + +1. **Create migration file:** + ```bash + migrate create -ext sql -dir migrations -seq add_new_column + ``` + +2. **Write up migration** (`migrations/XXXXXX_add_new_column.up.sql`): + ```sql + ALTER TABLE aviation.telegrams + ADD COLUMN new_field TEXT; + + CREATE INDEX idx_telegrams_new_field ON aviation.telegrams (new_field); + ``` + +3. **Write down migration** (`migrations/XXXXXX_add_new_column.down.sql`): + ```sql + DROP INDEX IF EXISTS idx_telegrams_new_field; + ALTER TABLE aviation.telegrams + DROP COLUMN IF EXISTS new_field; + ``` + +4. **Test migration:** + ```bash + # Apply + migrate -path migrations -database "$DATABASE_URL" up + + # Rollback + migrate -path migrations -database "$DATABASE_URL" down 1 + ``` + +### Production + +1. **Backup database** before applying migrations: + ```bash + pg_dump -U postgres -d aviation > backup_$(date +%Y%m%d_%H%M%S).sql + ``` + +2. **Test migration on staging** environment first + +3. **Apply migration** during maintenance window: + ```bash + migrate -path migrations -database "$DATABASE_URL" up + ``` + +4. **Verify migration** success: + ```bash + migrate -path migrations -database "$DATABASE_URL" version + ``` + +5. **Monitor application** for any issues + +## Schema Evolution Best Practices + +### 1. Backward Compatibility + +- **Additive changes** (new columns, indexes) are generally safe +- **Removing columns** requires application code changes first +- **Changing column types** requires careful planning and data migration + +### 2. TimescaleDB Considerations + +- **Hypertables**: Be careful when modifying hypertable structure +- **Retention policies**: Consider impact on existing data +- **Compression**: Test compression policies with schema changes + +### 3. Index Management + +- **Create indexes concurrently** in production to avoid locking: + ```sql + CREATE INDEX CONCURRENTLY idx_telegrams_new_field ON aviation.telegrams (new_field); + ``` + +- **Drop unused indexes** to improve write performance + +### 4. Data Migrations + +For data transformations, use separate migration steps: + +1. **Add new column** (nullable) +2. **Backfill data** in application or migration script +3. **Add constraints** (NOT NULL, etc.) after backfill +4. **Remove old column** in separate migration + +### 5. Rollback Procedures + +Always provide rollback migrations: + +- **Test rollback** on staging before production +- **Document rollback steps** in migration comments +- **Consider data loss** implications of rollbacks + +## Example Migration + +### Adding a New Index + +**Up migration:** +```sql +-- Add index for querying by category and date +CREATE INDEX CONCURRENTLY idx_telegrams_category_date +ON aviation.telegrams (category, received_at DESC); +``` + +**Down migration:** +```sql +-- Remove index +DROP INDEX IF EXISTS idx_telegrams_category_date; +``` + +### Adding a New Column + +**Up migration:** +```sql +-- Add processing_metadata column for additional metadata +ALTER TABLE aviation.telegrams +ADD COLUMN processing_metadata JSONB; + +-- Add index for JSONB queries +CREATE INDEX CONCURRENTLY idx_telegrams_processing_metadata_gin +ON aviation.telegrams USING GIN (processing_metadata); +``` + +**Down migration:** +```sql +-- Remove index and column +DROP INDEX IF EXISTS idx_telegrams_processing_metadata_gin; +ALTER TABLE aviation.telegrams +DROP COLUMN IF EXISTS processing_metadata; +``` + +## Migration State Management + +### Schema Version Tracking + +Migration tools typically use a `schema_migrations` table to track applied migrations: + +```sql +CREATE TABLE IF NOT EXISTS schema_migrations ( + version BIGINT NOT NULL PRIMARY KEY, + dirty BOOLEAN NOT NULL +); +``` + +### Checking Migration Status + +```bash +# Check current version +migrate -path migrations -database "$DATABASE_URL" version + +# Check for pending migrations +migrate -path migrations -database "$DATABASE_URL" up +``` + +## CI/CD Integration + +### Automated Migration Testing + +Add migration tests to CI pipeline: + +```yaml +# Example GitHub Actions workflow +- name: Test migrations + run: | + # Start test database + docker run -d -p 5432:5432 -e POSTGRES_PASSWORD=test postgres:15 + + # Wait for database + sleep 5 + + # Apply migrations + migrate -path migrations -database "postgres://postgres:test@localhost:5432/test?sslmode=disable" up + + # Verify schema + psql "postgres://postgres:test@localhost:5432/test?sslmode=disable" -c "\d aviation.telegrams" +``` + +### Deployment Automation + +For production deployments, integrate migrations into deployment process: + +1. **Pre-deployment**: Backup database +2. **Deployment**: Apply migrations +3. **Post-deployment**: Verify migration success +4. **Rollback**: If migration fails, rollback application and database + +## Troubleshooting + +### Migration Failures + +**Common issues:** +- **Lock conflicts**: Use `CONCURRENTLY` for index creation +- **Timeout errors**: Increase migration timeout for large tables +- **Dirty state**: Manually fix `schema_migrations` table if migration fails mid-way + +**Recovery:** +```sql +-- Check migration state +SELECT * FROM schema_migrations; + +-- Fix dirty state (if needed) +UPDATE schema_migrations SET dirty = false WHERE version = X; +``` + +### Performance Considerations + +- **Large tables**: Test migrations on production-sized data +- **Downtime**: Plan for maintenance windows for major schema changes +- **Replication lag**: Consider impact on read replicas + +## Future Improvements + +Consider implementing: + +1. **Automated migration testing** in CI/CD +2. **Migration rollback automation** in deployment pipeline +3. **Schema validation** before applying migrations +4. **Migration dry-run** mode for testing +5. **Migration status monitoring** and alerting + +## References + +- [golang-migrate Documentation](https://github.com/golang-migrate/migrate) +- [TimescaleDB Best Practices](https://docs.timescale.com/timescaledb/latest/how-to-guides/migrate-data/) +- [PostgreSQL Migration Guide](https://www.postgresql.org/docs/current/ddl-alter.html) + diff --git a/docs/nats.md b/docs/nats.md index 047d8d6..b1b1bd7 100644 --- a/docs/nats.md +++ b/docs/nats.md @@ -4,17 +4,45 @@ The NATS integration provides a robust, production-ready message processing system built on Clean Architecture principles. It supports both JetStream (persistent) and Core NATS (fire-and-forget) modes with comprehensive error handling, observability, and resilience features. +### Key Concepts + +1. **Consumer**: Pulls messages from NATS JetStream in batches, processes them, and handles ACKs/NAKs +2. **Publisher**: Publishes messages to NATS with automatic deduplication via UUID headers +3. **Batch Processing**: Fetches multiple messages at once (configurable size) for efficiency +4. **Error Classification**: Distinguishes between transient (retry) and permanent (DLQ) errors +5. **Dead Letter Queue (DLQ)**: Routes failed messages to a separate queue for analysis +6. **Backpressure**: Automatically slows down processing when errors accumulate +7. **Self-Healing**: Automatically recreates missing streams/consumers in development +8. **Observability**: Built-in metrics, tracing, and structured logging + +### Quick Start Flow + +``` +1. Configure NATS connection and consumer settings +2. Create Consumer with dependencies (processor, logger, telemetry) +3. Start Consumer - begins fetching and processing messages +4. Messages flow: Fetch → Process → ACK/NAK/DLQ +5. Errors handled automatically with retries and backoff +6. Graceful shutdown on context cancellation +``` + ## Architecture ### Clean Architecture Layers +The NATS integration follows Clean Architecture principles, separating concerns into distinct layers: + ``` ┌─────────────────────────────────────┐ │ Port Interfaces │ │ (Publisher, Consumer contracts) │ +│ - Define contracts, not impl │ +│ - Enable dependency inversion │ ├─────────────────────────────────────┤ │ Application Layer │ │ (Message processing logic) │ +│ - Business logic │ +│ - Use case orchestration │ ├─────────────────────────────────────┤ │ Infrastructure Layer │ │ (NATS implementation details) │ @@ -39,6 +67,337 @@ The NATS integration provides a robust, production-ready message processing syst └─────────────────────────────────────┘ ``` +### Component Interaction Diagram + +``` +┌──────────────┐ +│ Publisher │ +│ │ +│ 1. Serialize │ +│ 2. Add UUID │ +│ 3. Publish │ +└──────┬───────┘ + │ + │ Publish to Subject + ▼ +┌─────────────────────────────────────┐ +│ NATS JetStream │ +│ │ +│ ┌──────────────┐ │ +│ │ Stream │ │ +│ │ (TELEGRAM) │ │ +│ └──────┬───────┘ │ +│ │ │ +│ ┌──────▼───────┐ │ +│ │ Consumer │ │ +│ │ (Pull Sub) │ │ +│ └──────┬───────┘ │ +└─────────┼───────────────────────────┘ + │ + │ Fetch Batch + ▼ +┌─────────────────────────────────────┐ +│ Consumer │ +│ │ +│ ┌──────────────────────────────┐ │ +│ │ MessageFetcher │ │ +│ │ - FetchBatch() │ │ +│ │ - HandleFetchError() │ │ +│ └──────────┬───────────────────┘ │ +│ │ │ +│ ┌──────────▼───────────────────┐ │ +│ │ MessageProcessor │ │ +│ │ - ProcessBatch() │ │ +│ │ - ProcessSingleMessage() │ │ +│ └──────────┬───────────────────┘ │ +│ │ │ +│ ┌──────────▼───────────────────┐ │ +│ │ ErrorHandler │ │ +│ │ - Classify errors │ │ +│ │ - Apply backpressure │ │ +│ └──────────┬───────────────────┘ │ +│ │ │ +│ ┌──────────▼───────────────────┐ │ +│ │ DLQHandler │ │ +│ │ - RouteToDLQ() │ │ +│ │ - AdvisoryDLQHandler │ │ +│ └──────────────────────────────┘ │ +└─────────────────────────────────────┘ + │ + │ ACK/NAK + ▼ +┌─────────────────────────────────────┐ +│ Application Processor │ +│ (Business Logic) │ +└─────────────────────────────────────┘ +``` + +### Architecture Principles + +1. **Dependency Inversion**: High-level modules (Consumer, Publisher) depend on abstractions (interfaces), not concrete implementations +2. **Separation of Concerns**: Each component has a single responsibility: + - `MessageFetcher`: Handles message retrieval + - `MessageProcessor`: Handles message processing logic + - `ErrorHandler`: Handles error classification and recovery + - `DLQHandler`: Handles dead letter queue routing +3. **Testability**: All components can be mocked and tested independently +4. **Extensibility**: New implementations can be added without modifying existing code + +## Logic Flow + +### Consumer Processing Flow + +The consumer follows a well-defined processing loop with error handling at each stage: + +``` +┌─────────────────────────────────────────────────────────────┐ +│ Consumer Start │ +│ 1. Initialize components (Fetcher, Processor, DLQ) │ +│ 2. Create/validate JetStream resources │ +│ 3. Start advisory DLQ handler (if enabled) │ +│ 4. Start metrics collection goroutine │ +└──────────────────────┬──────────────────────────────────────┘ + │ + ▼ +┌─────────────────────────────────────────────────────────────┐ +│ Main Loop │ +│ │ +│ ┌──────────────────────────────────────────────────────┐ │ +│ │ Step 1: Check Context │ │ +│ │ - If cancelled, exit gracefully │ │ +│ └──────────────────┬───────────────────────────────────┘ │ +│ │ │ +│ ┌──────────────────▼───────────────────────────────────┐ │ +│ │ Step 2: Fetch Batch │ │ +│ │ - Fetch up to BatchSize messages │ │ +│ │ - Wait up to BatchTimeout │ │ +│ │ - Handle fetch errors with recovery │ │ +│ └──────────────────┬───────────────────────────────────┘ │ +│ │ │ +│ ┌───────────┴───────────┐ │ +│ │ │ │ +│ Success Error │ +│ │ │ │ +│ │ ┌────────▼────────┐ │ +│ │ │ Handle Error │ │ +│ │ │ - Classify type │ │ +│ │ │ - Apply backoff │ │ +│ │ │ - Recover if dev│ │ +│ │ └────────┬────────┘ │ +│ │ │ │ +│ │ ┌────────▼────────┐ │ +│ │ │ Continue? │ │ +│ │ └────────┬────────┘ │ +│ │ │ │ +│ │ Yes │ No │ +│ │ │ │ │ +│ │ └───┬────┴───┐ │ +│ │ │ │ │ +│ │ Continue Exit │ +│ │ │ │ +│ └──────────────────┘ │ +│ │ │ +│ ┌──────────────────▼───────────────────────────────────┐ │ +│ │ Step 3: Process Batch │ │ +│ │ - For each message in batch: │ │ +│ │ * Check context │ │ +│ │ * Extract message ID │ │ +│ │ * Create tracing span │ │ +│ │ * Call processor.Handle() │ │ +│ │ * Handle result (ACK/NAK/DLQ) │ │ +│ └──────────────────────────────────────────────────────┘ │ +│ │ │ +│ ┌───────────┴───────────┐ │ +│ │ │ │ +│ Success Error │ +│ │ │ │ +│ │ ┌────────▼────────┐ │ +│ │ │ Classify Error │ │ +│ │ └────────┬────────┘ │ +│ │ │ │ +│ │ ┌─────────────┴─────────────┐ │ +│ │ │ │ │ +│ │ Permanent Transient │ +│ │ │ │ │ +│ │ ┌──────▼──────┐ ┌────────▼──────┐ │ +│ │ │ Route to DLQ│ │ NAK with delay│ │ +│ │ │ ACK message │ │ Apply backpres│ │ +│ │ └──────┬──────┘ └────────┬──────┘ │ +│ │ │ │ │ +│ └─────────┴───────────────────────────┘ │ +│ │ │ +│ ┌──────────────────▼───────────────────────────────────┐ │ +│ │ Step 4: Reset Error Streak (if successful) │ │ +│ └──────────────────┬───────────────────────────────────┘ │ +│ │ │ +│ └─────────── Loop ─────────────────────┘ +└─────────────────────────────────────────────────────────────┘ +``` + +### Message Processing Logic + +#### Single Message Processing Flow + +``` +Message Received + │ + ▼ +┌─────────────────────┐ +│ Extract Message ID │ +│ - Check header │ +│ - Fallback to meta │ +│ - Generate if none │ +└──────────┬──────────┘ + │ + ▼ +┌─────────────────────┐ +│ Create Trace Span │ +│ - Add attributes │ +│ - Propagate context │ +└──────────┬──────────┘ + │ + ▼ +┌─────────────────────┐ +│ Process Message │ +│ - Call processor │ +│ - Business logic │ +└──────────┬──────────┘ + │ + ┌──────┴──────┐ + │ │ +Success Error + │ │ + │ ┌──────▼──────────┐ + │ │ Classify Error │ + │ └──────┬──────────┘ + │ │ + │ ┌──────┴──────┐ + │ │ │ + │ Permanent Transient + │ │ │ + │ ┌───▼───┐ ┌────▼────┐ + │ │ DLQ │ │ NAK │ + │ │ ACK │ │ Backoff │ + │ └───┬───┘ └────┬────┘ + │ │ │ + └──────┴─────────────┘ + │ + ▼ + End Processing +``` + +#### Error Handling Logic + +``` +Error Occurred + │ + ▼ +┌─────────────────────┐ +│ Is Permanent Error? │ +│ - app.IsPermanent() │ +└──────┬──────────────┘ + │ + ┌───┴───┐ + │ │ + Yes No + │ │ + │ ┌───▼──────────────────────┐ + │ │ Increment Error Streak │ + │ └───┬──────────────────────┘ + │ │ + │ ┌───▼──────────────────────┐ + │ │ Streak >= Threshold? │ + │ │ (default: 10 errors) │ + │ └───┬──────────────────────┘ + │ │ + │ ┌───┴───┐ + │ │ │ + │ Yes No + │ │ │ + │ │ ┌───▼──────────────┐ + │ │ │ NAK with delay │ + │ │ │ - Use backoff │ + │ │ │ - Request retry │ + │ │ └──────────────────┘ + │ │ + │ ▼ + │ ┌──────────────────────┐ + │ │ Apply Backpressure │ + │ │ - Sleep: errors*100ms│ + │ │ - Max: 5 seconds │ + │ └───┬──────────────────┘ + │ │ + │ ▼ + │ ┌──────────────────────┐ + │ │ NAK with delay │ + │ └──────────────────────┘ + │ + ▼ +┌──────────────────────┐ +│ Route to DLQ │ +│ - Enrich metadata │ +│ - Publish to DLQ │ +│ - ACK original msg │ +└──────────────────────┘ +``` + +### Publisher Logic Flow + +``` +Publish Request + │ + ▼ +┌─────────────────────┐ +│ Validate Topic │ +│ - Check config │ +└──────────┬──────────┘ + │ + ▼ +┌─────────────────────┐ +│ Serialize Message │ +│ - JSON marshal │ +└──────────┬──────────┘ + │ + ▼ +┌─────────────────────┐ +│ Extract/Generate ID │ +│ - From message.Uuid │ +│ - Or generate UUID │ +└──────────┬──────────┘ + │ + ▼ +┌─────────────────────┐ +│ Set Header │ +│ - Nats-Msg-Id │ +└──────────┬──────────┘ + │ + ▼ +┌─────────────────────┐ +│ Publish to NATS │ +│ - js.PublishMsg() │ +└──────────┬──────────┘ + │ + ┌──────┴──────┐ + │ │ +Success Error + │ │ + │ ┌──────▼──────────┐ + │ │ Classify Error │ + │ └──────┬──────────┘ + │ │ + │ ┌──────┴──────┐ + │ │ │ + │ Transient Permanent + │ │ │ + │ ┌───▼───┐ ┌────▼────┐ + │ │ Retry │ │ Fail │ + │ │ Later │ │ Fast │ + │ └───────┘ └─────────┘ + │ + ▼ +Success +``` + ## Core Components ### Consumer @@ -57,6 +416,38 @@ The consumer handles message consumption with the following features: - **Self-Healing**: Automatic recreation of missing streams/consumers in dev environments - **Graceful Shutdown**: Proper cleanup and draining of connections +#### Component Logic + +**MessageFetcher (`defaultMessageFetcher`)** +- Fetches batches of messages using `sub.Fetch(batchSize, MaxWait(timeout))` +- Handles fetch errors with exponential backoff +- Recovers subscriptions when connection issues occur +- Context-aware: respects cancellation signals + +**MessageProcessor (`defaultBatchProcessor`)** +- Processes messages sequentially within a batch +- Extracts message IDs (header → metadata → generated) +- Creates OpenTelemetry spans for tracing +- Calls application processor for business logic +- Handles ACK/NAK based on processing results + +**ErrorHandler** +- Classifies errors as transient or permanent using `app.IsPermanent()` +- Tracks consecutive error streaks +- Applies backpressure when streak exceeds threshold (default: 10) +- Calculates backoff delays for retries + +**DLQHandler (`defaultDLQHandler`)** +- Routes permanent errors to DLQ with enriched metadata +- Validates DLQ stream exists at startup +- Publishes DLQ messages with error context + +**AdvisoryDLQHandler** +- Subscribes to JetStream advisory events: `$JS.EVENT.ADVISORY.CONSUMER.MAX_DELIVERIES.*` +- Handles messages that exhaust MaxDeliver attempts +- Retrieves original message from stream using `GetMsg()` +- Routes to DLQ with advisory metadata + #### Configuration ```toml [NATS] @@ -92,6 +483,26 @@ The publisher handles message publishing with deduplication and observability. - **Structured Logging**: Comprehensive logging of publish operations - **Error Classification**: Distinguishes between transient and permanent errors +#### Component Logic + +**Publishing Flow** +1. **Validation**: Checks that publisher topic is configured +2. **Serialization**: Marshals message to JSON using `json.Marshal()` +3. **ID Extraction**: Extracts UUID from message (if `ParsedTelegram` type) or generates new UUID +4. **Header Attachment**: Sets `Nats-Msg-Id` header for deduplication +5. **Publishing**: Calls `js.PublishMsg()` to publish to JetStream +6. **Error Handling**: Classifies errors as transient (`ErrNoResponders`) or permanent + +**Deduplication Strategy** +- Uses `Nats-Msg-Id` header for JetStream deduplication +- Extracts UUID from `ParsedTelegram.Uuid` field if available +- Falls back to generating new UUID if not present +- JetStream uses this header to prevent duplicate message processing + +**Error Classification** +- **Transient**: `nats.ErrNoResponders` - JetStream temporarily unavailable, should retry +- **Permanent**: Other errors - configuration issues, should fail fast + ### Error Handling #### Error Types @@ -196,6 +607,294 @@ CAATSM_DLQ_SUBJECT=caatsm.dlq - **Performance**: Load testing and resource usage - **Configuration**: Different configuration combinations +## Simple Examples + +### Example 1: Complete Consumer Setup and Start + +This example shows how to set up and start a consumer from scratch: + +```go +package main + +import ( + "context" + "time" + + "caatsm/internal/infra/config" + "caatsm/internal/infra/nats" + "caatsm/internal/app" + "go.uber.org/zap" +) + +func main() { + // 1. Load configuration + cfg := &config.Config{ + NATS: config.NATSConfig{ + URL: "nats://localhost:4222", + Mode: "jetstream", + Stream: "TELEGRAM", + Consumer: "telegram-consumer", + ConsumerRules: config.ConsumerRules{ + AckWait: 30 * time.Second, + MaxDeliver: 3, + Backoff: []time.Duration{1*time.Second, 2*time.Second, 5*time.Second}, + }, + }, + App: config.AppConfig{ + BatchSize: 50, + BatchTimeout: 2 * time.Second, + }, + DLQ: config.DLQConfig{ + Enabled: true, + Subject: "caatsm.dlq", + }, + } + + // 2. Create NATS connection + nc, _ := nats.Connect(cfg.NATS.URL) + defer nc.Close() + + // 3. Get JetStream context + js, _ := nc.JetStream() + + // 4. Create message processor (your business logic) + processor := app.NewMessageProcessor(/* dependencies */) + + // 5. Create logger + logger, _ := zap.NewProduction() + + // 6. Create telemetry recorder + telemetry := /* your telemetry implementation */ + + // 7. Create consumer + consumer, err := natsinfra.ProvideConsumer( + nc, + js, + processor, + cfg, + telemetry, + logger, + ) + if err != nil { + logger.Fatal("Failed to create consumer", zap.Error(err)) + } + + // 8. Start consumer with context + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + // Handle graceful shutdown + go func() { + // Wait for interrupt signal + <-ctx.Done() + shutdownCtx, _ := context.WithTimeout(context.Background(), 5*time.Second) + consumer.Shutdown(shutdownCtx) + }() + + // 9. Start consuming (blocks until context cancelled) + if err := consumer.Start(ctx); err != nil { + logger.Error("Consumer stopped", zap.Error(err)) + } +} +``` + +### Example 2: Publishing a Message + +Simple example of publishing a message: + +```go +package main + +import ( + "caatsm/internal/adapter/dto" + "caatsm/internal/infra/nats" + "github.com/google/uuid" +) + +func publishMessage(publisher port.Publisher) error { + // Create message with UUID + message := &dto.ParsedTelegram{ + Uuid: uuid.NewString(), // Used for deduplication + Data: []byte("telegram message data"), + // ... other fields + } + + // Publish - automatically handles: + // - JSON serialization + // - UUID header attachment + // - Error classification + if err := publisher.Publish(message); err != nil { + return fmt.Errorf("failed to publish: %w", err) + } + + return nil +} +``` + +### Example 3: Message Processing Flow + +Step-by-step what happens when a message is processed: + +```go +// Step 1: Consumer fetches batch of messages +msgs, err := subscription.Fetch(50, nats.MaxWait(2*time.Second)) +// Result: Up to 50 messages, or timeout after 2 seconds + +// Step 2: For each message in batch +for _, msg := range msgs { + // Step 2a: Extract message ID + msgID := msg.Header.Get("Nats-Msg-Id") + if msgID == "" { + // Fallback: use JetStream sequence + meta, _ := msg.Metadata() + msgID = fmt.Sprintf("js-%d", meta.Sequence.Stream) + } + + // Step 2b: Create tracing span + ctx, span := tracer.Start(ctx, "process.message") + span.SetAttributes( + attribute.String("messaging.system", "nats"), + attribute.String("messaging.destination.name", msg.Subject), + ) + + // Step 2c: Process message (your business logic) + err := processor.Handle(ctx, msg.Data, msgID) + + // Step 2d: Handle result + if err != nil { + if app.IsPermanent(err) { + // Permanent error: route to DLQ and ACK + dlqHandler.RouteToDLQ(ctx, msg, err) + msg.Ack() + } else { + // Transient error: NAK with backoff + msg.NakWithDelay(calculateBackoff(msg)) + } + } else { + // Success: ACK message + msg.Ack() + } + + span.End() +} +``` + +### Example 4: Error Handling Scenarios + +Different error scenarios and how they're handled: + +```go +// Scenario 1: Transient Error (Network Issue) +func processMessage(msg *nats.Msg) error { + // Simulate network error + if networkDown { + return app.NewTransientError("network unavailable") + } + // Result: Message is NAK'd, will be redelivered with backoff +} + +// Scenario 2: Permanent Error (Invalid Format) +func processMessage(msg *nats.Msg) error { + var data MyStruct + if err := json.Unmarshal(msg.Data, &data); err != nil { + return app.NewPermanentError("invalid JSON format") + } + // Result: Message routed to DLQ, original message ACK'd +} + +// Scenario 3: Backpressure Trigger +// When 10+ consecutive errors occur: +// - Processing pauses +// - Sleep duration = min(consecutiveErrors * 100ms, 5s) +// - Prevents overwhelming the system + +// Scenario 4: MaxDeliver Exhausted +// When message fails MaxDeliver times (default: 3): +// - JetStream publishes advisory event +// - AdvisoryDLQHandler catches event +// - Retrieves original message +// - Routes to DLQ with metadata +``` + +### Example 5: DLQ Message Structure + +What a DLQ message looks like: + +```json +{ + "transport_msg_id": "550e8400-e29b-41d4-a716-446655440000", + "subject": "telegram.orders.12345", + "stream": "TELEGRAM", + "consumer": "telegram-consumer", + "nats_sequence": 12345, + "deliveries": 3, + "error": "permanent error: invalid message format", + "received_at": "2024-01-15T10:30:00Z", + "body": "{\"order_id\":123,\"invalid\":\"data\"}", + "advisory_source": false +} +``` + +### Example 6: Configuration Examples + +Different configuration scenarios: + +```toml +# Example 1: High Throughput Configuration +[NATS] +Mode = "jetstream" +Stream = "TELEGRAM" +Consumer = "telegram-consumer" + +[NATS.ConsumerRules] +AckWait = "60s" +MaxDeliver = 5 +MaxAckPending = 5000 +Backoff = ["1s", "2s", "5s", "10s", "30s"] + +[App] +BatchSize = 100 # Larger batches +BatchTimeout = "5s" # Longer timeout + +# Example 2: Low Latency Configuration +[App] +BatchSize = 10 # Smaller batches +BatchTimeout = "500ms" # Shorter timeout + +# Example 3: Development Mode (Self-Healing) +[NATS] +Mode = "jetstream" +# Missing streams/consumers auto-created + +# Example 4: Production Mode (Fail Fast) +[NATS] +Mode = "jetstream" +# Missing streams/consumers cause startup failure +``` + +### Example 7: Observability Integration + +How to monitor the consumer: + +```go +// Metrics are automatically collected: +// - ack_pending: Messages waiting for ACK +// - redelivered: Messages being redelivered +// - pending: Messages in stream +// - delivered: Total messages delivered + +// Tracing spans are created for: +// - Each message processing +// - DLQ routing +// - Error handling + +// Logs include: +// - Message processing events +// - Error details with context +// - DLQ routing events +// - Connection status changes +``` + ## Usage Examples ### Basic Consumer Setup diff --git a/docs/performance.md b/docs/performance.md new file mode 100644 index 0000000..f232687 --- /dev/null +++ b/docs/performance.md @@ -0,0 +1,396 @@ +# Performance Tuning Guide + +This document provides guidelines for optimizing the performance of the CAATSM application. + +## Performance Metrics + +Key performance indicators to monitor: + +- **Message throughput**: Messages processed per second +- **Latency**: End-to-end processing time (NATS receive → DB insert → publish) +- **Database query time**: Time spent on database operations +- **Memory usage**: Application memory consumption +- **CPU usage**: CPU utilization +- **Connection pool utilization**: Database and NATS connection usage + +## Batch Processing Configuration + +### Current Implementation + +The application processes messages in batches for efficiency: + +```toml +[app] +batch_size = 50 # Number of messages per batch +batch_timeout = "2s" # Maximum wait time for a batch +monitor_interval = "30s" # Consumer metrics reporting interval +``` + +### Tuning Guidelines + +**Batch Size:** +- **Small batches (10-50)**: Lower latency, higher overhead +- **Medium batches (50-200)**: Balanced latency and throughput +- **Large batches (200-1000)**: Higher throughput, higher latency + +**Recommendations:** +- **Development**: 10-50 messages (faster feedback) +- **Production (low latency)**: 50-100 messages +- **Production (high throughput)**: 200-500 messages + +**Batch Timeout:** +- **Low latency**: 500ms-1s (process quickly even with small batches) +- **Balanced**: 2-5s (good balance) +- **High throughput**: 5-10s (wait for larger batches) + +### Example Configuration + +```toml +# High-throughput production configuration +[app] +batch_size = 200 +batch_timeout = "5s" +monitor_interval = "30s" + +# Low-latency production configuration +[app] +batch_size = 50 +batch_timeout = "1s" +monitor_interval = "30s" +``` + +## Database Connection Pooling + +### Configuration + +```toml +[postgres] +url = "postgres://user:pass@db:5432/aviation?sslmode=require" +max_conns = 20 # Maximum connections in pool +min_conns = 5 # Minimum connections in pool +``` + +### Tuning Guidelines + +**Connection Pool Size:** +- **Formula**: `max_conns = (expected_concurrent_requests * avg_query_time) / target_latency` +- **Minimum**: 2-5 connections (small deployments) +- **Recommended**: 10-20 connections (medium deployments) +- **Maximum**: 50-100 connections (high-throughput deployments) + +**Considerations:** +- Each connection consumes memory (~2-5MB) +- PostgreSQL has a maximum connection limit (default: 100) +- Too many connections can degrade performance +- Use connection pooler (PgBouncer) for high concurrency + +### Example Configurations + +```toml +# Small deployment (single instance) +[postgres] +max_conns = 10 +min_conns = 2 + +# Medium deployment (2-3 instances) +[postgres] +max_conns = 20 +min_conns = 5 + +# Large deployment (5+ instances, use PgBouncer) +[postgres] +max_conns = 10 # Per instance +min_conns = 2 +# Use PgBouncer with pool_mode=transaction +``` + +### Connection Pool Monitoring + +Monitor connection pool metrics: +- Active connections +- Idle connections +- Connection wait time +- Connection errors + +## Database Indexing Strategy + +### Current Indexes + +The application creates indexes on key fields: + +```sql +CREATE INDEX idx_telegrams_message_id ON aviation.telegrams (message_id); +CREATE INDEX idx_telegrams_date_time ON aviation.telegrams (date_time); +CREATE INDEX idx_telegrams_priority_indicator ON aviation.telegrams (priority_indicator); +CREATE INDEX idx_telegrams_primary_address ON aviation.telegrams (primary_address); +CREATE INDEX idx_telegrams_received_at ON aviation.telegrams (received_at); +CREATE INDEX idx_telegrams_uuid ON aviation.telegrams (uuid); +``` + +### Index Optimization + +**Query Patterns:** +- **Time-range queries**: Index on `received_at` (already exists) +- **Message lookup**: Index on `message_id` (already exists) +- **Category filtering**: Consider index on `category` if frequently queried +- **Composite indexes**: For multi-column queries + +**Example Composite Index:** +```sql +-- For queries filtering by category and date range +CREATE INDEX idx_telegrams_category_received_at +ON aviation.telegrams (category, received_at DESC); +``` + +### Index Maintenance + +- **Monitor index usage**: Use `pg_stat_user_indexes` to identify unused indexes +- **Rebuild indexes**: Periodically rebuild indexes to reduce bloat +- **Concurrent creation**: Use `CREATE INDEX CONCURRENTLY` in production + +## NATS JetStream Performance Tuning + +### Stream Configuration + +```toml +[nats.stream_limits] +max_msgs = 1000000 # Maximum messages in stream +max_bytes = 1073741824 # Maximum size (1GB) +max_age = "168h" # Retention period (7 days) +discard = "old" # Discard policy +storage = "file" # Storage type (file or memory) +replicas = 3 # Number of replicas +``` + +### Tuning Guidelines + +**Storage Type:** +- **File storage**: Persistent, slower (recommended for production) +- **Memory storage**: Faster, ephemeral (suitable for high-throughput temporary streams) + +**Replicas:** +- **Single node**: 1 replica (development) +- **Production**: 3+ replicas (high availability) + +**Retention:** +- **Short retention**: Lower storage, faster cleanup +- **Long retention**: More storage, replay capability + +### Consumer Configuration + +```toml +[nats.consumer_rules] +max_deliver = 5 # Maximum redelivery attempts +ack_wait = "30s" # ACK wait time +max_ack_pending = 1024 # Maximum unacknowledged messages +deliver_policy = "new" # Delivery policy +backoff = ["5s", "30s", "2m"] # Retry delays +``` + +**Tuning:** +- **ack_wait**: Set based on processing time (processing_time * 2-3) +- **max_ack_pending**: Increase for high-throughput (1024-4096) +- **backoff**: Adjust based on failure patterns + +## Memory Optimization + +### Garbage Collection Tuning + +Set Go GC environment variables for production: + +```bash +# Balanced GC (default) +export GOGC=100 + +# Aggressive GC (lower memory, higher CPU) +export GOGC=50 + +# Conservative GC (higher memory, lower CPU) +export GOGC=200 +``` + +### Memory Profiling + +Use `pprof` to identify memory issues: + +```bash +# Enable memory profiling +go tool pprof http://localhost:2112/debug/pprof/heap + +# Generate memory profile +go tool pprof -alloc_space http://localhost:2112/debug/pprof/heap +``` + +## CPU Optimization + +### Goroutine Management + +- **Limit goroutines**: Use worker pools for concurrent processing +- **Context cancellation**: Properly cancel goroutines to prevent leaks +- **Monitor goroutine count**: Use `runtime.NumGoroutine()` + +### CPU Profiling + +```bash +# Enable CPU profiling +go tool pprof http://localhost:2112/debug/pprof/profile + +# 30-second CPU profile +go tool pprof http://localhost:2112/debug/pprof/profile?seconds=30 +``` + +## Monitoring and Profiling + +### Prometheus Metrics + +Key metrics to monitor: + +- `caatsm_messages_total`: Message throughput +- `caatsm_handle_latency_seconds`: Processing latency +- `caatsm_db_query_latency_seconds`: Database query time +- `caatsm_nats_consumer_pending_messages`: Consumer lag + +### Grafana Dashboards + +Create dashboards for: +- Message throughput over time +- Latency percentiles (P50, P95, P99) +- Error rates +- Resource utilization (CPU, memory, connections) + +### Profiling Endpoints + +The application exposes profiling endpoints (if enabled): + +```bash +# Heap profile +curl http://localhost:2112/debug/pprof/heap > heap.prof + +# CPU profile +curl http://localhost:2112/debug/pprof/profile?seconds=30 > cpu.prof + +# Goroutine profile +curl http://localhost:2112/debug/pprof/goroutine > goroutine.prof +``` + +## Performance Testing + +### Load Testing + +Use tools like `k6`, `wrk`, or `vegeta` for load testing: + +```bash +# Example: Generate load with seed-telegrams +go run ./cmd/seed-telegrams \ + --count=10000 \ + --mode=burst \ + --category=mixed +``` + +### Benchmark Tests + +Run built-in benchmarks: + +```bash +# Run all benchmarks +go test -bench=. -benchmem ./... + +# Run specific benchmark +go test -bench=BenchmarkParser -benchmem ./internal/adapter/parser +``` + +### Performance Baselines + +Establish performance baselines: +- **Throughput**: Messages per second +- **Latency**: P50, P95, P99 percentiles +- **Resource usage**: CPU, memory, connections + +## Optimization Checklist + +### Application Level + +- [ ] Optimize batch size for workload +- [ ] Tune connection pool sizes +- [ ] Review and optimize database queries +- [ ] Add missing indexes for query patterns +- [ ] Enable query result caching where appropriate + +### Infrastructure Level + +- [ ] Use connection pooler (PgBouncer) for high concurrency +- [ ] Configure database connection limits appropriately +- [ ] Use read replicas for query-heavy workloads +- [ ] Optimize NATS JetStream stream configuration +- [ ] Scale horizontally (multiple instances) + +### Monitoring + +- [ ] Set up performance dashboards +- [ ] Configure alerts for performance degradation +- [ ] Regular performance profiling +- [ ] Monitor resource utilization +- [ ] Track performance trends over time + +## Troubleshooting Performance Issues + +### High Latency + +**Symptoms:** +- Slow message processing +- High P95/P99 latencies + +**Investigation:** +1. Check database query times +2. Review NATS consumer lag +3. Profile CPU and memory usage +4. Check for connection pool exhaustion + +**Solutions:** +- Optimize slow database queries +- Increase batch size +- Add database indexes +- Scale horizontally + +### Low Throughput + +**Symptoms:** +- Low messages per second +- High CPU usage + +**Investigation:** +1. Check for bottlenecks (DB, NATS, CPU) +2. Review batch processing configuration +3. Profile application code + +**Solutions:** +- Increase batch size +- Optimize hot code paths +- Scale horizontally +- Use connection pooling + +### High Memory Usage + +**Symptoms:** +- Memory leaks +- High memory consumption + +**Investigation:** +1. Heap profiling +2. Check for goroutine leaks +3. Review batch sizes + +**Solutions:** +- Fix memory leaks +- Reduce batch sizes +- Tune GC settings +- Limit concurrent operations + +## References + +- [Go Performance Best Practices](https://go.dev/doc/effective_go#performance) +- [PostgreSQL Performance Tuning](https://www.postgresql.org/docs/current/performance-tips.html) +- [TimescaleDB Performance Tuning](https://docs.timescale.com/timescaledb/latest/how-to-guides/performance/) +- [NATS JetStream Performance](https://docs.nats.io/nats-concepts/jetstream/performance) +- [Go Profiling Guide](https://go.dev/blog/pprof) + diff --git a/docs/prod-guide.md b/docs/prod-guide.md index e417266..3b2e2f3 100644 --- a/docs/prod-guide.md +++ b/docs/prod-guide.md @@ -421,6 +421,146 @@ backoff = ["5s", "30s", "2m", "5m"] # Retry delays - Don't log message payloads in production - Use appropriate log levels +## NATS Authentication + +The application supports multiple NATS authentication methods for secure connections. Configure authentication in the `[nats.auth]` section of your production configuration. + +### Authentication Methods + +Only one authentication method can be used at a time. Choose the method that best fits your infrastructure: + +#### 1. Token Authentication + +Simple token-based authentication suitable for service-to-service communication: + +```toml +[nats.auth] +token = "your-nats-token-here" +tls_enabled = true +``` + +**When to use:** +- Simple service-to-service authentication +- Single token shared across services +- Quick setup for development/staging + +**Security considerations:** +- Tokens should be rotated regularly +- Store tokens securely (use secret management) +- Use TLS to encrypt token transmission + +#### 2. Credentials File (Recommended) + +NATS credentials file authentication provides fine-grained access control: + +```toml +[nats.auth] +credentials_file = "/etc/caatsm/nats.creds" +tls_enabled = true +``` + +**When to use:** +- Production environments requiring fine-grained permissions +- Multiple services with different access levels +- Integration with NATS account system + +**Setup:** +1. Generate credentials file using NATS CLI: + ```bash + nats account creds -n caatsm-service > /etc/caatsm/nats.creds + ``` +2. Ensure the file is readable by the application user +3. Set appropriate file permissions (e.g., `chmod 600 /etc/caatsm/nats.creds`) + +#### 3. User/Password Authentication + +Traditional username/password authentication: + +```toml +[nats.auth] +user = "caatsm-service" +password = "secure-password-here" +tls_enabled = true +``` + +**When to use:** +- Legacy NATS server configurations +- Simple authentication requirements +- Integration with existing user management systems + +**Security considerations:** +- Use strong, unique passwords +- Store passwords securely (use secret management) +- Rotate passwords regularly + +### TLS Configuration + +TLS encryption is **required** for production deployments. Configure TLS in the `[nats.auth]` section: + +```toml +[nats.auth] +credentials_file = "/etc/caatsm/nats.creds" +tls_enabled = true +tls_cert_file = "/etc/caatsm/tls/client.crt" # Optional: client certificate +tls_key_file = "/etc/caatsm/tls/client.key" # Optional: client private key +tls_ca_file = "/etc/caatsm/tls/ca.crt" # Optional: CA certificate for server verification +``` + +**TLS Options:** +- `tls_enabled`: Enable TLS encryption (required for production) +- `tls_cert_file`: Client certificate file path (for mutual TLS) +- `tls_key_file`: Client private key file path (for mutual TLS) +- `tls_ca_file`: CA certificate file for server certificate verification + +**Note:** If `tls_ca_file` is not specified, the system's default CA certificates are used. For production, it's recommended to specify a CA file for explicit certificate validation. The application will load and use the CA certificate file for server verification when provided. + +### Environment Variable Configuration + +You can also configure authentication via environment variables: + +```bash +# Token authentication +export CAATSM_NATS_AUTH_TOKEN="your-token" + +# Credentials file +export CAATSM_NATS_AUTH_CREDENTIALS_FILE="/etc/caatsm/nats.creds" + +# User/Password +export CAATSM_NATS_AUTH_USER="caatsm-service" +export CAATSM_NATS_AUTH_PASSWORD="secure-password" + +# TLS +export CAATSM_NATS_AUTH_TLS_ENABLED="true" +export CAATSM_NATS_AUTH_TLS_CERT_FILE="/etc/caatsm/tls/client.crt" +export CAATSM_NATS_AUTH_TLS_KEY_FILE="/etc/caatsm/tls/client.key" +export CAATSM_NATS_AUTH_TLS_CA_FILE="/etc/caatsm/tls/ca.crt" +``` + +### Testing Authentication + +After configuring authentication, verify the connection: + +```bash +# Test connection with authentication +./bin/receiver listen --nats-url nats://nats.prod:4222 + +# Check logs for authentication success +# Look for: "NATS reconnected" or connection errors +``` + +### Troubleshooting + +**Connection failures:** +- Verify authentication credentials are correct +- Check NATS server logs for authentication errors +- Ensure TLS certificates are valid and accessible +- Verify file permissions on credentials/certificate files + +**Common errors:** +- `authentication failed`: Check token/credentials/user-password +- `tls: bad certificate`: Verify TLS certificate configuration +- `permission denied`: Check file permissions on credentials/certificate files + ## Backup and Recovery ### Database Backups diff --git a/docs/secret-management.md b/docs/secret-management.md new file mode 100644 index 0000000..0d71bb4 --- /dev/null +++ b/docs/secret-management.md @@ -0,0 +1,427 @@ +# Secret Management Guide + +This document describes best practices for managing secrets and sensitive configuration in the CAATSM application. + +## Current Approach + +The application currently supports secrets via environment variables with the `CAATSM_` prefix: + +```bash +export CAATSM_POSTGRES_URL="postgres://user:password@localhost:5432/aviation" +export CAATSM_NATS_AUTH_TOKEN="your-token" +export CAATSM_NATS_AUTH_PASSWORD="secure-password" +``` + +**Security considerations:** +- Environment variables are visible to all processes on the system +- Secrets may be logged in process lists or shell history +- No automatic rotation or expiration +- Manual management required + +## Recommended Secret Management Systems + +For production deployments, use a dedicated secret management system: + +### Option 1: HashiCorp Vault (Recommended) + +[Vault](https://www.vaultproject.io/) provides secure secret storage with dynamic secrets, encryption, and access control. + +#### Setup + +1. **Install Vault:** + ```bash + # Download and install Vault + wget https://releases.hashicorp.com/vault/1.15.0/vault_1.15.0_linux_amd64.zip + unzip vault_1.15.0_linux_amd64.zip + sudo mv vault /usr/local/bin/ + ``` + +2. **Start Vault (dev mode for testing):** + ```bash + vault server -dev + ``` + +3. **Store secrets:** + ```bash + export VAULT_ADDR='http://127.0.0.1:8200' + vault kv put secret/caatsm \ + postgres_url="postgres://user:pass@db:5432/aviation" \ + nats_token="your-token" \ + nats_password="secure-password" + ``` + +#### Integration + +Create a wrapper script or init container to fetch secrets from Vault: + +```bash +#!/bin/bash +# fetch-secrets.sh + +export VAULT_ADDR="${VAULT_ADDR:-http://vault:8200}" +export VAULT_TOKEN="${VAULT_TOKEN}" + +# Fetch secrets from Vault +vault kv get -format=json secret/caatsm | jq -r '.data.data | to_entries | .[] | "export CAATSM_\(.key | ascii_upcase | gsub("-"; "_"))=\(.value)"' > /tmp/secrets.env + +# Source secrets +source /tmp/secrets.env + +# Start application +exec ./bin/receiver listen +``` + +#### Kubernetes Integration + +Use Vault Agent Sidecar or Vault Secrets Operator: + +```yaml +apiVersion: v1 +kind: Pod +metadata: + name: caatsm-receiver +spec: + containers: + - name: vault-agent + image: vault:latest + command: ["/bin/sh", "-c"] + args: + - | + vault agent -config=/vault/config/agent.hcl + - name: caatsm-receiver + image: caatsm/receiver:latest + envFrom: + - secretRef: + name: caatsm-secrets +``` + +### Option 2: AWS Secrets Manager + +For AWS deployments, use [AWS Secrets Manager](https://aws.amazon.com/secrets-manager/) for centralized secret management. + +#### Setup + +1. **Store secrets:** + ```bash + aws secretsmanager create-secret \ + --name caatsm/production \ + --secret-string '{ + "postgres_url": "postgres://user:pass@db:5432/aviation", + "nats_token": "your-token", + "nats_password": "secure-password" + }' + ``` + +2. **Retrieve secrets:** + ```bash + aws secretsmanager get-secret-value \ + --secret-id caatsm/production \ + --query SecretString \ + --output text | jq -r 'to_entries | .[] | "export CAATSM_\(.key | ascii_upcase | gsub("-"; "_"))=\(.value)"' + ``` + +#### Integration + +Use AWS SDK or CLI in init container: + +```bash +#!/bin/bash +# fetch-aws-secrets.sh + +SECRET_JSON=$(aws secretsmanager get-secret-value \ + --secret-id caatsm/production \ + --query SecretString \ + --output text) + +echo "$SECRET_JSON" | jq -r 'to_entries | .[] | "export CAATSM_\(.key | ascii_upcase | gsub("-"; "_"))=\(.value)"' > /tmp/secrets.env + +source /tmp/secrets.env +exec ./bin/receiver listen +``` + +### Option 3: Kubernetes Secrets + +For Kubernetes deployments, use [Kubernetes Secrets](https://kubernetes.io/docs/concepts/configuration/secret/). + +#### Setup + +1. **Create secret:** + ```bash + kubectl create secret generic caatsm-secrets \ + --from-literal=postgres-url="postgres://user:pass@db:5432/aviation" \ + --from-literal=nats-token="your-token" \ + --from-literal=nats-password="secure-password" + ``` + +2. **Use in deployment:** + ```yaml + apiVersion: apps/v1 + kind: Deployment + metadata: + name: caatsm-receiver + spec: + template: + spec: + containers: + - name: receiver + image: caatsm/receiver:latest + env: + - name: CAATSM_POSTGRES_URL + valueFrom: + secretKeyRef: + name: caatsm-secrets + key: postgres-url + - name: CAATSM_NATS_AUTH_TOKEN + valueFrom: + secretKeyRef: + name: caatsm-secrets + key: nats-token + ``` + +#### Best Practices + +- **Encrypt at rest**: Enable encryption for etcd (Kubernetes backend) +- **RBAC**: Restrict access to secrets using Role-Based Access Control +- **External Secrets Operator**: Use [External Secrets Operator](https://external-secrets.io/) for integration with external secret stores + +### Option 4: Docker Secrets + +For Docker Swarm deployments, use [Docker Secrets](https://docs.docker.com/engine/swarm/secrets/). + +#### Setup + +1. **Create secret:** + ```bash + echo "your-secret-value" | docker secret create caatsm_nats_token - + ``` + +2. **Use in service:** + ```yaml + version: '3.8' + services: + receiver: + image: caatsm/receiver:latest + secrets: + - caatsm_nats_token + environment: + - CAATSM_NATS_AUTH_TOKEN_FILE=/run/secrets/caatsm_nats_token + ``` + +## Secret Rotation + +### Manual Rotation + +1. **Update secret** in secret management system +2. **Restart application** to pick up new secret +3. **Verify** application is working correctly +4. **Remove old secret** after verification + +### Automated Rotation + +For AWS Secrets Manager, enable automatic rotation: + +```bash +aws secretsmanager rotate-secret \ + --secret-id caatsm/production \ + --rotation-lambda-arn arn:aws:lambda:region:account:function:rotate-secret +``` + +For Vault, use dynamic secrets or scheduled rotation policies. + +## Security Best Practices + +### 1. Principle of Least Privilege + +- **Minimal access**: Grant only necessary permissions +- **Service accounts**: Use dedicated service accounts for applications +- **Secret scoping**: Limit secrets to specific services/environments + +### 2. Encryption + +- **Encryption at rest**: Ensure secrets are encrypted in storage +- **Encryption in transit**: Use TLS for secret retrieval +- **Key management**: Use proper key management (HSM, KMS, etc.) + +### 3. Audit and Monitoring + +- **Audit logs**: Enable audit logging for secret access +- **Monitoring**: Monitor secret access patterns +- **Alerts**: Set up alerts for unusual access patterns + +### 4. Secret Lifecycle + +- **Rotation**: Rotate secrets regularly (e.g., every 90 days) +- **Expiration**: Set expiration dates for secrets +- **Revocation**: Have a process for revoking compromised secrets + +### 5. Development vs Production + +- **Separate stores**: Use different secret stores for dev/staging/prod +- **No production secrets in code**: Never commit production secrets +- **Local development**: Use local secret files or dev vault instance + +## Configuration Examples + +### Environment Variables (Current) + +```bash +# Development +export CAATSM_POSTGRES_URL="postgres://user:pass@localhost:5432/aviation?sslmode=disable" +export CAATSM_NATS_URL="nats://localhost:4222" +export CAATSM_NATS_AUTH_TOKEN="dev-token" + +# Production (via secret management) +# Secrets loaded from Vault/AWS/K8s before application start +``` + +### Configuration File (Not Recommended for Secrets) + +```toml +# config.prod.toml +# DO NOT store secrets in config files +# Use environment variables or secret management instead + +[postgres] +# URL should come from CAATSM_POSTGRES_URL env var +url = "" # Empty, will be overridden by env var + +[nats.auth] +# Token should come from CAATSM_NATS_AUTH_TOKEN env var +token = "" # Empty, will be overridden by env var +``` + +## Secret Injection Patterns + +### Pattern 1: Init Container (Kubernetes) + +```yaml +apiVersion: v1 +kind: Pod +metadata: + name: caatsm-receiver +spec: + initContainers: + - name: fetch-secrets + image: vault:latest + command: ["/bin/sh", "-c"] + args: + - | + vault kv get -format=json secret/caatsm | \ + jq -r '.data.data | to_entries | .[] | "\(.key | ascii_upcase | gsub("-"; "_"))=\(.value)"' > \ + /shared/secrets.env + volumeMounts: + - name: shared-secrets + mountPath: /shared + containers: + - name: receiver + image: caatsm/receiver:latest + envFrom: + - configMapRef: + name: caatsm-config + env: + - name: CAATSM_SECRETS_FILE + value: /shared/secrets.env + volumeMounts: + - name: shared-secrets + mountPath: /shared + volumes: + - name: shared-secrets + emptyDir: {} +``` + +### Pattern 2: Sidecar Container + +```yaml +apiVersion: v1 +kind: Pod +metadata: + name: caatsm-receiver +spec: + containers: + - name: vault-agent + image: vault:latest + command: ["vault", "agent", "-config=/vault/config/agent.hcl"] + volumeMounts: + - name: vault-config + mountPath: /vault/config + - name: receiver + image: caatsm/receiver:latest + envFrom: + - secretRef: + name: caatsm-secrets + volumes: + - name: vault-config + configMap: + name: vault-agent-config +``` + +### Pattern 3: Application-Level Integration + +For applications that need to fetch secrets at runtime: + +```go +// Example: Fetch secrets from Vault at startup +func loadSecretsFromVault() error { + client, err := vault.NewClient(vault.DefaultConfig()) + if err != nil { + return err + } + + secret, err := client.Logical().Read("secret/data/caatsm") + if err != nil { + return err + } + + // Set environment variables + for k, v := range secret.Data["data"].(map[string]interface{}) { + os.Setenv("CAATSM_"+strings.ToUpper(k), v.(string)) + } + + return nil +} +``` + +## Troubleshooting + +### Secret Not Found + +**Symptoms:** +- Application fails to start +- Connection errors to database/NATS + +**Solutions:** +- Verify secret exists in secret store +- Check secret name/path is correct +- Verify application has permissions to access secret +- Check secret format (JSON, plain text, etc.) + +### Secret Access Denied + +**Symptoms:** +- Authentication errors when fetching secrets +- Permission denied errors + +**Solutions:** +- Verify IAM roles/service accounts have correct permissions +- Check Vault policies or AWS IAM policies +- Verify authentication tokens/credentials are valid + +### Secret Rotation Issues + +**Symptoms:** +- Application fails after secret rotation +- Connection errors after rotation + +**Solutions:** +- Implement graceful secret reloading +- Use connection pooling with automatic reconnection +- Test rotation process in staging first + +## References + +- [HashiCorp Vault Documentation](https://www.vaultproject.io/docs) +- [AWS Secrets Manager Documentation](https://docs.aws.amazon.com/secretsmanager/) +- [Kubernetes Secrets Documentation](https://kubernetes.io/docs/concepts/configuration/secret/) +- [External Secrets Operator](https://external-secrets.io/) +- [12-Factor App: Config](https://12factor.net/config) + diff --git a/go.mod b/go.mod index 2ac8812..e2d051f 100644 --- a/go.mod +++ b/go.mod @@ -1,6 +1,6 @@ module caatsm -go 1.25.0 +go 1.24.0 require ( github.com/google/uuid v1.6.0 diff --git a/internal/adapter/parser/aviation_parser_bench_test.go b/internal/adapter/parser/aviation_parser_bench_test.go new file mode 100644 index 0000000..1b91d53 --- /dev/null +++ b/internal/adapter/parser/aviation_parser_bench_test.go @@ -0,0 +1,137 @@ +package parser + +import ( + "testing" +) + +// Sample messages for benchmarking +var ( + benchARRMessage = `ZCZC TMQ2526 141605 +FF ZBTJZPZX +141604 ZBACZQZX +(ARR-JAE7433/A0132-RKSI-ZBTJ1604) +NNNN` + + benchDEPMessage = `ZCZC DEP5678 120915 +DD KLAXZPZX +120914 KSFOZQZX +(DEP-ABC5678-A1234-ZBTJ1440-ZGGG) +NNNN` + + benchCNLMessage = `ZCZC CNL9012 150631 +FF ZBTJZPZX +(CNL-CCA9012-ZBTJ-ZGGG) +NNNN` + + benchDLAMessage = `ZCZC DLA3456 150631 +FF ZBTJZPZX +(DLA-CCA3456-A1234-ZBTJ1600-ZGGG0200) +NNNN` + + benchFPLMessage = `ZCZC TMQ2617 142150 +GG ZBTJZPZX +150551 ZBTJUOBK +(FPL-OKA2861-IS +-MA60/M-SHID/C +-ZBTJ0030 +-K0420S0450 CG J1 FZ +-ZSYT0100 ZSQD ZYTL +-REG/B3710 SEL/ RMK/TCAS ) +NNNN` + + benchComplexFPLMessage = `ZCZC FPL7890 150631 +FF ZBTJZPZX +(FPL-JAE7433-IS +-B744/H-SXIRPZJWY/S +-ZBTJ1755 +-K0926S0920 CG A326 VYK W80 HUR B339 GM A575 MANSA/K0919S0980 +-EDDF0948 EDDK +-EET/ZMUB0100 UNKL0236 +REG/B2422 SEL/JLAD +NAV/RNAV1 RNAV5 RNP4 +RMK/AGCS EQUIPPED) +NNNN` +) + +// BenchmarkParseARR benchmarks parsing ARR messages +func BenchmarkParseARR(b *testing.B) { + b.ResetTimer() + for i := 0; i < b.N; i++ { + _, _ = Parse(benchARRMessage) + } +} + +// BenchmarkParseDEP benchmarks parsing DEP messages +func BenchmarkParseDEP(b *testing.B) { + b.ResetTimer() + for i := 0; i < b.N; i++ { + _, _ = Parse(benchDEPMessage) + } +} + +// BenchmarkParseCNL benchmarks parsing CNL messages +func BenchmarkParseCNL(b *testing.B) { + b.ResetTimer() + for i := 0; i < b.N; i++ { + _, _ = Parse(benchCNLMessage) + } +} + +// BenchmarkParseDLA benchmarks parsing DLA messages +func BenchmarkParseDLA(b *testing.B) { + b.ResetTimer() + for i := 0; i < b.N; i++ { + _, _ = Parse(benchDLAMessage) + } +} + +// BenchmarkParseFPL benchmarks parsing simple FPL messages +func BenchmarkParseFPL(b *testing.B) { + b.ResetTimer() + for i := 0; i < b.N; i++ { + _, _ = Parse(benchFPLMessage) + } +} + +// BenchmarkParseComplexFPL benchmarks parsing complex FPL messages with extensive route and metadata +func BenchmarkParseComplexFPL(b *testing.B) { + b.ResetTimer() + for i := 0; i < b.N; i++ { + _, _ = Parse(benchComplexFPLMessage) + } +} + +// BenchmarkParseHeader benchmarks header parsing only +func BenchmarkParseHeader(b *testing.B) { + b.ResetTimer() + for i := 0; i < b.N; i++ { + _, _ = ParseHeader(benchARRMessage) + } +} + +// BenchmarkParseBody benchmarks body parsing only (ARR) +func BenchmarkParseBody(b *testing.B) { + body := `(ARR-JAE7433/A0132-RKSI-ZBTJ1604)` + parser := NewBodyParser(body) + b.ResetTimer() + for i := 0; i < b.N; i++ { + _, _, _ = parser.Parse() + } +} + +// BenchmarkParseMixed benchmarks parsing a mix of message types +func BenchmarkParseMixed(b *testing.B) { + messages := []string{ + benchARRMessage, + benchDEPMessage, + benchCNLMessage, + benchDLAMessage, + benchFPLMessage, + } + b.ResetTimer() + for i := 0; i < b.N; i++ { + msg := messages[i%len(messages)] + _, _ = Parse(msg) + } +} + diff --git a/internal/app/processor_bench_test.go b/internal/app/processor_bench_test.go new file mode 100644 index 0000000..1357dd6 --- /dev/null +++ b/internal/app/processor_bench_test.go @@ -0,0 +1,147 @@ +package app + +import ( + "caatsm/internal/adapter/dto" + "caatsm/internal/adapter/parser" + "caatsm/internal/infra/telemetry" + "context" + "testing" + + "go.uber.org/zap" +) + +// Mock implementations for benchmarking +type mockRepository struct{} + +func (m *mockRepository) InsertOne(ctx context.Context, msg *dto.ParsedTelegram) error { + return nil +} + +func (m *mockRepository) InsertBatch(ctx context.Context, msgs []*dto.ParsedTelegram) error { + return nil +} + +func (m *mockRepository) InsertRaw(ctx context.Context, msg *dto.ParsedTelegram) error { + return nil +} + +type mockPublisher struct{} + +func (m *mockPublisher) Publish(message interface{}) error { + return nil +} + +// Benchmark data +var ( + benchARRRaw = []byte(`ZCZC TMQ2526 141605 +FF ZBTJZPZX +141604 ZBACZQZX +(ARR-JAE7433/A0132-RKSI-ZBTJ1604) +NNNN`) + + benchDEPRaw = []byte(`ZCZC DEP5678 120915 +DD KLAXZPZX +120914 KSFOZQZX +(DEP-ABC5678-A1234-ZBTJ1440-ZGGG) +NNNN`) + + benchFPLRaw = []byte(`ZCZC TMQ2617 142150 +GG ZBTJZPZX +150551 ZBTJUOBK +(FPL-OKA2861-IS +-MA60/M-SHID/C +-ZBTJ0030 +-K0420S0450 CG J1 FZ +-ZSYT0100 ZSQD ZYTL +-REG/B3710 SEL/ RMK/TCAS ) +NNNN`) +) + +// createBenchmarkProcessor creates a processor with mocks for benchmarking +func createBenchmarkProcessor() *MessageProcessor { + aviationParser := parser.ProvideParser() + mockRepo := &mockRepository{} + mockPub := &mockPublisher{} + logger := zap.NewNop() + recorder := telemetry.NewNoop() + + return NewMessageProcessor(aviationParser, mockRepo, mockPub, recorder, logger) +} + +// BenchmarkHandleARR benchmarks processing ARR messages end-to-end +func BenchmarkHandleARR(b *testing.B) { + processor := createBenchmarkProcessor() + ctx := context.Background() + + b.ResetTimer() + b.ReportAllocs() + + for i := 0; i < b.N; i++ { + _ = processor.Handle(ctx, benchARRRaw, "msg-arr-123") + } +} + +// BenchmarkHandleDEP benchmarks processing DEP messages end-to-end +func BenchmarkHandleDEP(b *testing.B) { + processor := createBenchmarkProcessor() + ctx := context.Background() + + b.ResetTimer() + b.ReportAllocs() + + for i := 0; i < b.N; i++ { + _ = processor.Handle(ctx, benchDEPRaw, "msg-dep-123") + } +} + +// BenchmarkHandleFPL benchmarks processing FPL messages end-to-end +func BenchmarkHandleFPL(b *testing.B) { + processor := createBenchmarkProcessor() + ctx := context.Background() + + b.ResetTimer() + b.ReportAllocs() + + for i := 0; i < b.N; i++ { + _ = processor.Handle(ctx, benchFPLRaw, "msg-fpl-123") + } +} + +// BenchmarkHandleMixed benchmarks processing a mix of message types +func BenchmarkHandleMixed(b *testing.B) { + processor := createBenchmarkProcessor() + ctx := context.Background() + messages := [][]byte{benchARRRaw, benchDEPRaw, benchFPLRaw} + msgIDs := []string{"msg-arr", "msg-dep", "msg-fpl"} + + b.ResetTimer() + b.ReportAllocs() + + for i := 0; i < b.N; i++ { + idx := i % len(messages) + _ = processor.Handle(ctx, messages[idx], msgIDs[idx]) + } +} + +// BenchmarkHandleParseOnly benchmarks parsing without persistence/publishing +// This isolates parser performance +func BenchmarkHandleParseOnly(b *testing.B) { + aviationParser := parser.ProvideParser() + // Use a repository that does nothing + mockRepo := &mockRepository{} + // Use a publisher that does nothing + mockPub := &mockPublisher{} + logger := zap.NewNop() + recorder := telemetry.NewNoop() + + processor := NewMessageProcessor(aviationParser, mockRepo, mockPub, recorder, logger) + ctx := context.Background() + + b.ResetTimer() + b.ReportAllocs() + + for i := 0; i < b.N; i++ { + _ = processor.Handle(ctx, benchARRRaw, "msg-parse-only") + } +} + diff --git a/internal/infra/config/config.go b/internal/infra/config/config.go index 32be97a..a55471d 100644 --- a/internal/infra/config/config.go +++ b/internal/infra/config/config.go @@ -35,11 +35,32 @@ type NATSConfig struct { Consumer string `koanf:"consumer"` StreamLimits StreamLimitsConfig `koanf:"stream_limits"` ConsumerRules ConsumerRulesConfig `koanf:"consumer_rules"` + Auth NATSAuthConfig `koanf:"auth"` // Legacy fields Client string `koanf:"client"` Cluster string `koanf:"cluster"` } +// NATSAuthConfig holds NATS authentication configuration +type NATSAuthConfig struct { + // Token authentication (mutually exclusive with User/Password and CredentialsFile) + Token string `koanf:"token"` + + // Credentials file authentication (mutually exclusive with Token and User/Password) + // Path to NATS credentials file (e.g., /path/to/user.creds) + CredentialsFile string `koanf:"credentials_file"` + + // User/Password authentication (mutually exclusive with Token and CredentialsFile) + User string `koanf:"user"` + Password string `koanf:"password"` + + // TLS configuration + TLSEnabled bool `koanf:"tls_enabled"` + TLSCertFile string `koanf:"tls_cert_file"` // Client certificate file + TLSKeyFile string `koanf:"tls_key_file"` // Client private key file + TLSCAFile string `koanf:"tls_ca_file"` // CA certificate file for server verification +} + // StreamLimitsConfig defines JetStream retention controls. type StreamLimitsConfig struct { MaxMsgs int64 `koanf:"max_msgs"` @@ -173,8 +194,9 @@ func LoadConfig() (*Config, error) { return strings.ToLower(strings.ReplaceAll(s, "_", ".")) }) if err := k.Load(envProvider, nil); err != nil { - // Environment variables are optional, so we don't fail if they're not present - // This allows the config to work with just the file + // Environment variables are optional, so we don't fail if they're not present. + // This allows the config to work with just the file. + _ = err // explicitly ignore } // Unmarshal into Config struct diff --git a/internal/infra/log/logger_test.go b/internal/infra/log/logger_test.go index a30f06a..a4b7136 100644 --- a/internal/infra/log/logger_test.go +++ b/internal/infra/log/logger_test.go @@ -37,7 +37,12 @@ var _ = Describe("ProvideLogger", func() { Context("when file output is configured", func() { It("creates the directory before writing logs", func() { tmpDir := filepath.Join(os.TempDir(), "caatsm-log-test") - defer os.RemoveAll(tmpDir) + defer func() { + if err := os.RemoveAll(tmpDir); err != nil { + // Cleanup errors in tests are not critical + _ = err + } + }() logPath := filepath.Join(tmpDir, "child", "app.log") cfg := &configpkg.Config{ Log: configpkg.LogConfig{ diff --git a/internal/infra/nats/advisory_dlq.go b/internal/infra/nats/advisory_dlq.go index 69d34e1..b826c8e 100644 --- a/internal/infra/nats/advisory_dlq.go +++ b/internal/infra/nats/advisory_dlq.go @@ -85,7 +85,9 @@ func (h *AdvisoryDLQHandler) Start(ctx context.Context) error { // Wait for context cancellation go func() { <-ctx.Done() - sub.Unsubscribe() + if err := sub.Unsubscribe(); err != nil { + h.logger.Error("Failed to unsubscribe advisory subscription", zap.Error(err)) + } h.logger.Info("Stopped advisory DLQ handler") }() diff --git a/internal/infra/nats/consumer.go b/internal/infra/nats/consumer.go index 49ece60..55caefb 100644 --- a/internal/infra/nats/consumer.go +++ b/internal/infra/nats/consumer.go @@ -3,6 +3,8 @@ package nats import ( "caatsm/internal/app" "caatsm/internal/infra/config" + "caatsm/internal/infra/log" + obsmetrics "caatsm/internal/infra/metrics" "caatsm/internal/infra/telemetry" "context" "encoding/json" @@ -11,7 +13,11 @@ import ( "strings" "time" + "github.com/google/uuid" "github.com/nats-io/nats.go" + "go.opentelemetry.io/otel" + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/codes" "go.opentelemetry.io/otel/metric" "go.uber.org/zap" ) @@ -167,7 +173,9 @@ func (f *defaultMessageFetcher) HandleFetchError(ctx context.Context, err error, ) // Connection closed is fatal - cannot recover subscription if *sub != nil { - (*sub).Unsubscribe() + if err := (*sub).Unsubscribe(); err != nil { + f.logger.Error("Failed to unsubscribe after connection closed", zap.Error(err)) + } *sub = nil } return false, fmt.Errorf("connection closed: %w", err) @@ -209,7 +217,9 @@ func (f *defaultMessageFetcher) HandleFetchError(ctx context.Context, err error, } // Unsubscribe old subscription before creating new one if *sub != nil { - (*sub).Unsubscribe() + if err := (*sub).Unsubscribe(); err != nil { + f.logger.Error("Failed to unsubscribe during recovery", zap.Error(err)) + } } // Create new subscription newSub, subErr := f.consumerManager.CreatePullSubscription() @@ -229,7 +239,9 @@ func (f *defaultMessageFetcher) HandleFetchError(ctx context.Context, err error, zap.String("consumer", f.config.consumerName), ) if *sub != nil { - (*sub).Unsubscribe() + if err := (*sub).Unsubscribe(); err != nil { + f.logger.Error("Failed to unsubscribe after resource not found", zap.Error(err)) + } *sub = nil } return false, fmt.Errorf("JetStream resource not found: %w", err) @@ -322,7 +334,9 @@ func (f *defaultMessageFetcher) attemptSubscriptionRecovery(ctx context.Context, // Unsubscribe old subscription if it exists if *sub != nil { - (*sub).Unsubscribe() + if err := (*sub).Unsubscribe(); err != nil { + f.logger.Error("Failed to unsubscribe during recovery", zap.Error(err)) + } *sub = nil } @@ -368,10 +382,238 @@ type defaultBatchProcessor struct { errorHandler *ErrorHandler logger *zap.Logger telemetry telemetry.Recorder + // Configuration needed for processing + streamName string + consumerName string + mode string + backoff []time.Duration + // Pointer to consecutive errors counter (shared with Consumer) + consecutiveProcessErrors *int } func (p *defaultBatchProcessor) ProcessBatch(ctx context.Context, msgs []*nats.Msg) { - // This will be implemented when we refactor the batch processing + for _, msg := range msgs { + // Check context before processing each message + select { + case <-ctx.Done(): + p.logger.Info("Stopping batch processing due to cancellation", + zap.Int("remaining_messages", len(msgs)), + ) + return + default: + } + p.processSingleMessage(ctx, msg) + } +} + +// processSingleMessage processes a single message with error handling and backpressure. +func (p *defaultBatchProcessor) processSingleMessage(ctx context.Context, msg *nats.Msg) { + start := time.Now() + + if err := p.processMessage(ctx, msg); err != nil { + p.handleMessageError(ctx, msg, err, time.Since(start)) + return + } + + // Successful processing resets the error streak. + if p.consecutiveProcessErrors != nil && *p.consecutiveProcessErrors > 0 { + *p.consecutiveProcessErrors = 0 + } + + // ACK the message + if ackErr := msg.Ack(); ackErr != nil { + p.logger.Error("Failed to ACK message", zap.Error(ackErr)) + } else { + elapsed := time.Since(start) + p.telemetry.RecordMessageHandled(ctx, p.streamName, p.consumerName, "ok", elapsed) + } +} + +// processMessage processes a single message. +func (p *defaultBatchProcessor) processMessage(ctx context.Context, msg *nats.Msg) error { + ctx, span := otel.Tracer("caatsm/nats").Start(ctx, "Consumer.processMessage") + defer span.End() + + // Set semantic messaging attributes + span.SetAttributes( + attribute.String("messaging.system", "nats"), + attribute.String("messaging.operation.name", "receive"), + attribute.String("messaging.destination.name", msg.Subject), + attribute.String("messaging.consumer.group.name", p.consumerName), + attribute.String("caatsm.stream", p.streamName), + ) + + msgID, source, err := p.resolveMsgID(msg) + if err != nil { + span.RecordError(err) + span.SetStatus(codes.Error, err.Error()) + return fmt.Errorf("unable to resolve message id: %w", err) + } + if source != "header" { + p.logger.Warn("Message missing NATS id header; using fallback", + zap.String("subject", msg.Subject), + zap.String("msg_id_source", source), + zap.String("msg_id", msgID), + ) + } + + // Attach structured logging context including stream/consumer and NATS metadata. + jsSeq := uint64(0) + if meta, metaErr := msg.Metadata(); metaErr == nil { + jsSeq = meta.Sequence.Stream + span.SetAttributes( + attribute.Int64("nats.js.stream_seq", int64(meta.Sequence.Stream)), + attribute.Int64("nats.js.consumer_seq", int64(meta.Sequence.Consumer)), + ) + } + + msgLogger := log.WithMessageContext(p.logger, log.MessageFields{ + Service: "caatsm-consumer", + TransportMsgID: msgID, + Stream: p.streamName, + Consumer: p.consumerName, + Subject: msg.Subject, + JSSequence: jsSeq, + }) + + msgLogger.Debug("Processing message", + zap.Int("data_size", len(msg.Data)), + zap.String("msg_id_source", source), + ) + + // Call processor + if err := p.processor.Handle(ctx, msg.Data, msgID); err != nil { + span.RecordError(err) + span.SetStatus(codes.Error, err.Error()) + return fmt.Errorf("processor error: %w", err) + } + + span.SetAttributes(attribute.String("telegram.msg_id", msgID)) + return nil +} + +// resolveMsgID extracts or generates a message ID. +func (p *defaultBatchProcessor) resolveMsgID(msg *nats.Msg) (string, string, error) { + if id := msg.Header.Get("Nats-Msg-Id"); id != "" { + return id, "header", nil + } + + if p.mode == "core" { + return uuid.NewString(), "generated", nil + } + + meta, err := msg.Metadata() + if err != nil { + return "", "", fmt.Errorf("fetch metadata: %w", err) + } + + return fmt.Sprintf("js-%d", meta.Sequence.Stream), "metadata", nil +} + +// handleMessageError handles errors that occur during message processing. +func (p *defaultBatchProcessor) handleMessageError(ctx context.Context, msg *nats.Msg, err error, elapsed time.Duration) { + p.logger.Error("Failed to process message", + zap.String("subject", msg.Subject), + zap.Error(err), + zap.Bool("permanent", app.IsPermanent(err)), + ) + + result := obsmetrics.ResultFail + if app.IsPermanent(err) { + result = obsmetrics.ResultPermanentFail + } + p.telemetry.RecordMessageHandled(ctx, p.streamName, p.consumerName, result, elapsed) + + consecutiveErrors := 0 + if p.consecutiveProcessErrors != nil { + consecutiveErrors = *p.consecutiveProcessErrors + } + + processingResult := p.errorHandler.HandleProcessingError(consecutiveErrors, err, p.logger, msg.Subject) + + if processingResult.IsPermanent { + p.handlePermanentError(ctx, msg, err) + return + } + + p.handleTransientError(ctx, msg, processingResult) +} + +// handlePermanentError handles permanent/poison messages. +func (p *defaultBatchProcessor) handlePermanentError(ctx context.Context, msg *nats.Msg, err error) { + if p.consecutiveProcessErrors != nil { + *p.consecutiveProcessErrors = 0 + } + // Poison/permanent message: route to DLQ if configured, then ACK + if p.dlqHandler != nil { + if dlqErr := p.dlqHandler.RouteToDLQ(ctx, msg, err); dlqErr != nil { + p.logger.Error("Failed to route permanent-error message to DLQ", zap.Error(dlqErr)) + } + } + if ackErr := msg.Ack(); ackErr != nil { + p.logger.Error("Failed to ACK permanent-error message", zap.Error(ackErr)) + } +} + +// handleTransientError handles transient errors with backpressure and redelivery. +func (p *defaultBatchProcessor) handleTransientError(ctx context.Context, msg *nats.Msg, processingResult ProcessingErrorResult) { + // Increment error streak + if p.consecutiveProcessErrors != nil { + if *p.consecutiveProcessErrors < 0 { + *p.consecutiveProcessErrors = 0 + } + *p.consecutiveProcessErrors++ + } + + if processingResult.ShouldApplyBackpressure { + consecutiveErrors := 0 + if p.consecutiveProcessErrors != nil { + consecutiveErrors = *p.consecutiveProcessErrors + } + p.logger.Warn("Applying backpressure due to consecutive processing errors", + zap.Int("consecutive_errors", consecutiveErrors), + zap.Duration("sleep", processingResult.BackpressureDelay), + ) + // Use context-aware sleep instead of blocking time.Sleep + if !sleepWithContext(ctx, processingResult.BackpressureDelay) { + // Context canceled, stop processing + return + } + } + + // Transient error: request redelivery with optional delay + p.telemetry.RecordRetry(ctx, p.streamName, p.consumerName, obsmetrics.RetryReasonProcessorError) + if nakErr := p.nakWithStrategy(msg); nakErr != nil { + p.logger.Error("Failed to NAK message", zap.Error(nakErr)) + } +} + +// nakWithStrategy sends a NAK with appropriate delay based on retry attempt. +func (p *defaultBatchProcessor) nakWithStrategy(msg *nats.Msg) error { + if len(p.backoff) == 0 { + return msg.Nak() + } + + meta, err := msg.Metadata() + if err != nil { + p.logger.Warn("Failed to read metadata for backoff strategy", zap.Error(err)) + return msg.Nak() + } + + attempt := int(meta.NumDelivered) + index := attempt - 1 + if index < 0 { + index = 0 + } + if index >= len(p.backoff) { + index = len(p.backoff) - 1 + } + delay := p.backoff[index] + if delay <= 0 { + return msg.Nak() + } + + return msg.NakWithDelay(delay) } // defaultDLQHandler implements DLQHandler interface @@ -454,14 +696,7 @@ func (c *Consumer) initCollaborators() { cfg: c.cfg, } - c.batchProcessor = &defaultBatchProcessor{ - processor: c.processor, - dlqHandler: c.dlqHandler, - errorHandler: c.errorHandler, - logger: c.logger, - telemetry: c.telemetry, - } - + // Initialize DLQ handler first if needed, so batch processor can reference it if c.config.dlqSubject != "" { c.dlqHandler = &defaultDLQHandler{ js: c.js, @@ -472,6 +707,19 @@ func (c *Consumer) initCollaborators() { telemetry: c.telemetry, } } + + c.batchProcessor = &defaultBatchProcessor{ + processor: c.processor, + dlqHandler: c.dlqHandler, + errorHandler: c.errorHandler, + logger: c.logger, + telemetry: c.telemetry, + streamName: c.config.streamName, + consumerName: c.config.consumerName, + mode: c.config.mode, + backoff: c.cfg.NATS.ConsumerRules.Backoff, + consecutiveProcessErrors: &c.consecutiveProcessErrors, + } } // normalizeConsumerConfig extracts and normalizes consumer configuration from the application config. diff --git a/internal/infra/nats/consumer_error.go b/internal/infra/nats/consumer_error.go index 2377b19..3caf7f9 100644 --- a/internal/infra/nats/consumer_error.go +++ b/internal/infra/nats/consumer_error.go @@ -11,6 +11,8 @@ import ( ) // handleMessageError handles errors that occur during message processing. +// +//nolint:unused // Reserved for potential future use or alternative implementation func (c *Consumer) handleMessageError(ctx context.Context, msg *nats.Msg, err error, elapsed time.Duration) { c.logger.Error("Failed to process message", zap.String("subject", msg.Subject), @@ -35,6 +37,8 @@ func (c *Consumer) handleMessageError(ctx context.Context, msg *nats.Msg, err er } // handlePermanentError handles permanent/poison messages. +// +//nolint:unused // Reserved for potential future use or alternative implementation func (c *Consumer) handlePermanentError(ctx context.Context, msg *nats.Msg, err error) { c.consecutiveProcessErrors = 0 // Poison/permanent message: route to DLQ if configured, then ACK @@ -47,6 +51,8 @@ func (c *Consumer) handlePermanentError(ctx context.Context, msg *nats.Msg, err } // handleTransientError handles transient errors with backpressure and redelivery. +// +//nolint:unused // Reserved for potential future use or alternative implementation func (c *Consumer) handleTransientError(ctx context.Context, msg *nats.Msg, processingResult ProcessingErrorResult) { // Increment error streak if c.consecutiveProcessErrors < 0 { diff --git a/internal/infra/nats/consumer_jetstream.go b/internal/infra/nats/consumer_jetstream.go index 42dce73..64824f9 100644 --- a/internal/infra/nats/consumer_jetstream.go +++ b/internal/infra/nats/consumer_jetstream.go @@ -32,6 +32,7 @@ func (c *Consumer) createPullSubscriptionWithRecovery() (*nats.Subscription, err return c.consumerManager.CreatePullSubscriptionWithRecovery(c.streamManager, consumerConfig) } +//nolint:unused // Reserved for potential future use or alternative implementation // nakWithStrategy sends a NAK with appropriate delay based on retry attempt. func (c *Consumer) nakWithStrategy(msg *nats.Msg) error { backoff := c.cfg.NATS.ConsumerRules.Backoff @@ -74,6 +75,7 @@ func sleepWithContext(ctx context.Context, duration time.Duration) bool { } } +//nolint:unused // Reserved for potential future use or alternative implementation // fetchBatch fetches a batch of messages from the subscription. // It respects context cancellation for faster shutdown. func (c *Consumer) fetchBatch(ctx context.Context, sub *nats.Subscription) ([]*nats.Msg, error) { @@ -102,7 +104,11 @@ func (c *Consumer) handleFetchError(ctx context.Context, err error, sub **nats.S if recErr := c.recoverJetStreamResources(); recErr != nil { return nil, recErr } - (*sub).Unsubscribe() + if *sub != nil { + if unsubErr := (*sub).Unsubscribe(); unsubErr != nil { + c.logger.Error("Failed to unsubscribe during recovery", zap.Error(unsubErr)) + } + } return c.createPullSubscriptionWithRecovery() }) @@ -125,10 +131,12 @@ func (c *Consumer) startJetStream(ctx context.Context) error { // Use a closure that always cleans up the current subscription. // When subscription is replaced in handleFetchError, this will clean up // whatever currentSub points to at shutdown time. - var currentSub *nats.Subscription = sub + var currentSub = sub cleanupSubscriber := func() { if currentSub != nil { - currentSub.Unsubscribe() + if err := currentSub.Unsubscribe(); err != nil { + c.logger.Error("Failed to unsubscribe subscription", zap.Error(err)) + } currentSub = nil } } @@ -174,14 +182,14 @@ func (c *Consumer) startJetStream(ctx context.Context) error { } // Fetch messages in batch - msgs, err := c.fetchBatch(ctx, currentSub) + msgs, err := c.fetcher.FetchBatch(ctx, currentSub) if err != nil { // If context was cancelled, return immediately if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) { c.logger.Info("Stopping consumer due to context cancellation", zap.Error(err)) return err } - shouldContinue, handleErr := c.handleFetchError(ctx, err, ¤tSub, &fetchErrorStreak) + shouldContinue, handleErr := c.fetcher.HandleFetchError(ctx, err, ¤tSub, &fetchErrorStreak) if !shouldContinue { return handleErr } @@ -194,6 +202,6 @@ func (c *Consumer) startJetStream(ctx context.Context) error { } // Process batch - c.processBatch(ctx, msgs) + c.batchProcessor.ProcessBatch(ctx, msgs) } } diff --git a/internal/infra/nats/consumer_message.go b/internal/infra/nats/consumer_message.go index 81d5790..ed55a3f 100644 --- a/internal/infra/nats/consumer_message.go +++ b/internal/infra/nats/consumer_message.go @@ -14,6 +14,7 @@ import ( "go.uber.org/zap" ) +//nolint:unused // Reserved for potential future use or alternative implementation // processBatch processes a batch of messages, handling errors and applying backpressure. // It checks context cancellation between messages for faster shutdown. func (c *Consumer) processBatch(ctx context.Context, msgs []*nats.Msg) { @@ -31,6 +32,7 @@ func (c *Consumer) processBatch(ctx context.Context, msgs []*nats.Msg) { } } +//nolint:unused // Reserved for potential future use or alternative implementation // processSingleMessage processes a single message with error handling and backpressure. func (c *Consumer) processSingleMessage(ctx context.Context, msg *nats.Msg) { start := time.Now() diff --git a/internal/infra/nats/jetstream.go b/internal/infra/nats/jetstream.go index a6411b9..025e7ed 100644 --- a/internal/infra/nats/jetstream.go +++ b/internal/infra/nats/jetstream.go @@ -2,17 +2,19 @@ package nats import ( "caatsm/internal/infra/config" + "crypto/tls" + "crypto/x509" "fmt" + "os" "strings" "github.com/nats-io/nats.go" "go.uber.org/zap" ) -// ProvideNATSConn creates a reusable NATS connection. +// ProvideNATSConn creates a reusable NATS connection with optional authentication. func ProvideNATSConn(cfg *config.Config, logger *zap.Logger) (*nats.Conn, error) { - nc, err := nats.Connect( - cfg.NATS.URL, + opts := []nats.Option{ nats.RetryOnFailedConnect(true), nats.Timeout(cfg.Timeouts.Server), nats.ReconnectWait(cfg.Timeouts.ReconnectWait), @@ -27,7 +29,16 @@ func ProvideNATSConn(cfg *config.Config, logger *zap.Logger) (*nats.Conn, error) safeURL := sanitizeURLForLogging(nc.ConnectedUrl()) logger.Info("NATS reconnected", zap.String("url", safeURL)) }), - ) + } + + // Apply authentication options + authOpts, err := buildAuthOptions(&cfg.NATS.Auth, logger) + if err != nil { + return nil, fmt.Errorf("failed to build auth options: %w", err) + } + opts = append(opts, authOpts...) + + nc, err := nats.Connect(cfg.NATS.URL, opts...) if err != nil { safeURL := sanitizeURLForLogging(cfg.NATS.URL) logger.Error("failed to connect to NATS", @@ -42,6 +53,78 @@ func ProvideNATSConn(cfg *config.Config, logger *zap.Logger) (*nats.Conn, error) return nc, nil } +// buildAuthOptions builds NATS connection options based on authentication configuration. +func buildAuthOptions(auth *config.NATSAuthConfig, logger *zap.Logger) ([]nats.Option, error) { + var opts []nats.Option + authMethods := 0 + + // Token authentication (highest priority) + if auth.Token != "" { + authMethods++ + logger.Debug("Using NATS token authentication") + opts = append(opts, nats.Token(auth.Token)) + } + + // Credentials file authentication + if auth.CredentialsFile != "" { + authMethods++ + if authMethods > 1 { + return nil, fmt.Errorf("multiple authentication methods specified: only one of token, credentials_file, or user/password can be used") + } + logger.Debug("Using NATS credentials file authentication", zap.String("file", auth.CredentialsFile)) + opts = append(opts, nats.UserCredentials(auth.CredentialsFile)) + } + + // User/Password authentication + if auth.User != "" || auth.Password != "" { + authMethods++ + if authMethods > 1 { + return nil, fmt.Errorf("multiple authentication methods specified: only one of token, credentials_file, or user/password can be used") + } + if auth.User == "" || auth.Password == "" { + return nil, fmt.Errorf("both user and password must be specified for user/password authentication") + } + logger.Debug("Using NATS user/password authentication", zap.String("user", auth.User)) + opts = append(opts, nats.UserInfo(auth.User, auth.Password)) + } + + // TLS configuration + if auth.TLSEnabled { + tlsConfig := &tls.Config{ + MinVersion: tls.VersionTLS12, + } + + // Load client certificate and key if provided + if auth.TLSCertFile != "" && auth.TLSKeyFile != "" { + cert, err := tls.LoadX509KeyPair(auth.TLSCertFile, auth.TLSKeyFile) + if err != nil { + return nil, fmt.Errorf("failed to load TLS certificate: %w", err) + } + tlsConfig.Certificates = []tls.Certificate{cert} + logger.Debug("Loaded TLS client certificate", zap.String("cert", auth.TLSCertFile)) + } + + // Load CA certificate for server verification if provided + if auth.TLSCAFile != "" { + caCert, err := os.ReadFile(auth.TLSCAFile) + if err != nil { + return nil, fmt.Errorf("failed to read CA certificate file: %w", err) + } + caCertPool := x509.NewCertPool() + if !caCertPool.AppendCertsFromPEM(caCert) { + return nil, fmt.Errorf("failed to parse CA certificate from %s", auth.TLSCAFile) + } + tlsConfig.RootCAs = caCertPool + logger.Debug("Loaded TLS CA certificate", zap.String("ca_file", auth.TLSCAFile)) + } + + opts = append(opts, nats.Secure(tlsConfig)) + logger.Debug("TLS enabled for NATS connection") + } + + return opts, nil +} + // ProvideJetStream creates a NATS JetStream context using an existing connection. // Returns nil, nil when cfg.NATS.Mode == "core" to support plain NATS servers without JetStream. func ProvideJetStream(nc *nats.Conn, cfg *config.Config, logger *zap.Logger) (nats.JetStreamContext, error) { diff --git a/internal/infra/nats/utils_test.go b/internal/infra/nats/utils_test.go index 799cfe4..fb13be4 100644 --- a/internal/infra/nats/utils_test.go +++ b/internal/infra/nats/utils_test.go @@ -14,45 +14,72 @@ var _ = Describe("Utils", func() { originalEnv := os.Getenv("GO_ENV") DeferCleanup(func() { if originalEnv == "" { - os.Unsetenv("GO_ENV") + if err := os.Unsetenv("GO_ENV"); err != nil { + // Environment variables are optional, ignore cleanup errors in tests + _ = err + } } else { - os.Setenv("GO_ENV", originalEnv) + if err := os.Setenv("GO_ENV", originalEnv); err != nil { + // Environment variables are optional, ignore cleanup errors in tests + _ = err + } } }) }) It("returns true for dev environment", func() { - os.Setenv("GO_ENV", "dev") + if err := os.Setenv("GO_ENV", "dev"); err != nil { + // Environment variables are optional, ignore setup errors in tests + _ = err + } Expect(isDevLikeEnv()).To(BeTrue()) }) It("returns true for development environment", func() { - os.Setenv("GO_ENV", "development") + if err := os.Setenv("GO_ENV", "development"); err != nil { + // Environment variables are optional, ignore setup errors in tests + _ = err + } Expect(isDevLikeEnv()).To(BeTrue()) }) It("returns true for test environment", func() { - os.Setenv("GO_ENV", "test") + if err := os.Setenv("GO_ENV", "test"); err != nil { + // Environment variables are optional, ignore setup errors in tests + _ = err + } Expect(isDevLikeEnv()).To(BeTrue()) }) It("returns true for testing environment", func() { - os.Setenv("GO_ENV", "testing") + if err := os.Setenv("GO_ENV", "testing"); err != nil { + // Environment variables are optional, ignore setup errors in tests + _ = err + } Expect(isDevLikeEnv()).To(BeTrue()) }) It("returns true for empty environment", func() { - os.Unsetenv("GO_ENV") + if err := os.Unsetenv("GO_ENV"); err != nil { + // Environment variables are optional, ignore cleanup errors in tests + _ = err + } Expect(isDevLikeEnv()).To(BeTrue()) }) It("returns false for production environment", func() { - os.Setenv("GO_ENV", "prod") + if err := os.Setenv("GO_ENV", "prod"); err != nil { + // Environment variables are optional, ignore setup errors in tests + _ = err + } Expect(isDevLikeEnv()).To(BeFalse()) }) It("returns false for production environment (uppercase)", func() { - os.Setenv("GO_ENV", "PROD") + if err := os.Setenv("GO_ENV", "PROD"); err != nil { + // Environment variables are optional, ignore setup errors in tests + _ = err + } Expect(isDevLikeEnv()).To(BeFalse()) }) }) diff --git a/internal/infra/postgres/repository.go b/internal/infra/postgres/repository.go index c2a24b7..8066b46 100644 --- a/internal/infra/postgres/repository.go +++ b/internal/infra/postgres/repository.go @@ -37,6 +37,10 @@ func ProvideRepository(pool *pgxpool.Pool, logger *zap.Logger) (port.Repository, // InsertOne inserts a single telegram message func (r *Repository) InsertOne(ctx context.Context, msg *dto.ParsedTelegram) error { + if msg == nil { + return fmt.Errorf("message cannot be nil") + } + ctx, span := otel.Tracer("caatsm/postgres").Start(ctx, "Repository.InsertOne") defer span.End() @@ -53,7 +57,7 @@ func (r *Repository) InsertOne(ctx context.Context, msg *dto.ParsedTelegram) err // Optional idempotency check based on business message identity. If we have a // non-empty message ID and date/time, we can cheaply skip duplicates here to // avoid applying the same business event multiple times. - if msg != nil && msg.MessageID != "" && msg.DateTime != "" { + if msg.MessageID != "" && msg.DateTime != "" { exists, err := r.messageExists(ctx, msg.MessageID, msg.DateTime) if err != nil { span.RecordError(err) diff --git a/internal/infra/postgres/repository_bench_test.go b/internal/infra/postgres/repository_bench_test.go new file mode 100644 index 0000000..ecc61a8 --- /dev/null +++ b/internal/infra/postgres/repository_bench_test.go @@ -0,0 +1,169 @@ +package postgres + +import ( + "caatsm/internal/adapter/dto" + "caatsm/internal/adapter/mapper" + "context" + "strconv" + "testing" + "time" + + "github.com/google/uuid" +) + +// Benchmark data setup +func createBenchmarkTelegram() *dto.ParsedTelegram { + return &dto.ParsedTelegram{ + Uuid: uuid.New().String(), + MessageID: "TMQ1234", + DateTime: "150631", + PriorityIndicator: "FF", + PrimaryAddress: "ZBTJZPZX", + Category: "ARR", + Content: "ZCZC TMQ1234 150631\nFF ZBTJZPZX\n(ARR-ABC123-ZBTJ-ZGGG)\nNNNN", + Status: dto.MessageStatusParsed, + Parsed: true, + ReceivedAt: time.Now(), + ParsedAt: time.Now(), + } +} + +func createBenchmarkTelegrams(count int) []*dto.ParsedTelegram { + telegrams := make([]*dto.ParsedTelegram, count) + for i := 0; i < count; i++ { + tg := createBenchmarkTelegram() + tg.Uuid = uuid.New().String() + tg.MessageID = "TMQ" + strconv.Itoa(1000+i) + telegrams[i] = tg + } + return telegrams +} + +// BenchmarkInsertOne benchmarks single message insertion +// Note: This requires a database connection. Run with -tags=integration or provide test DB. +func BenchmarkInsertOne(b *testing.B) { + // Skip if no database available (integration tests only) + if testing.Short() { + b.Skip("Skipping benchmark in short mode") + } + + // This benchmark requires a real database connection + // In practice, you would set up a test database connection here + // For now, we'll skip if not running integration tests + b.Skip("Requires database connection - run with integration tests") +} + +// BenchmarkInsertOneWithDB benchmarks single message insertion with database +// This is a helper that can be used in integration test suites +// +//nolint:unused // Benchmark helper for future use +func benchmarkInsertOneWithDB(b *testing.B, repo *Repository) { + ctx := context.Background() + telegram := createBenchmarkTelegram() + + b.ResetTimer() + b.ReportAllocs() + + for i := 0; i < b.N; i++ { + // Update UUID for each iteration to avoid conflicts + telegram.Uuid = uuid.New().String() + telegram.MessageID = "TMQ" + strconv.Itoa(1000+i) + _ = repo.InsertOne(ctx, telegram) + } +} + +// BenchmarkInsertBatch benchmarks batch message insertion +// Note: This requires a database connection. Run with -tags=integration or provide test DB. +func BenchmarkInsertBatch(b *testing.B) { + // Skip if no database available (integration tests only) + if testing.Short() { + b.Skip("Skipping benchmark in short mode") + } + + // This benchmark requires a real database connection + // In practice, you would set up a test database connection here + // For now, we'll skip if not running integration tests + b.Skip("Requires database connection - run with integration tests") +} + +// BenchmarkInsertBatchWithDB benchmarks batch insertion with database +// This is a helper that can be used in integration test suites +// +//nolint:unused // Benchmark helper for future use +func benchmarkInsertBatchWithDB(b *testing.B, repo *Repository, batchSize int) { + ctx := context.Background() + telegrams := createBenchmarkTelegrams(batchSize) + + b.ResetTimer() + b.ReportAllocs() + + for i := 0; i < b.N; i++ { + // Update UUIDs for each iteration to avoid conflicts + for j := range telegrams { + telegrams[j].Uuid = uuid.New().String() + telegrams[j].MessageID = "TMQ" + strconv.Itoa(1000+i*batchSize+j) + } + _ = repo.InsertBatch(ctx, telegrams) + } +} + +// BenchmarkInsertBatch10 benchmarks batch insertion with 10 messages +func BenchmarkInsertBatch10(b *testing.B) { + if testing.Short() { + b.Skip("Skipping benchmark in short mode") + } + b.Skip("Requires database connection - run with integration tests") +} + +// BenchmarkInsertBatch50 benchmarks batch insertion with 50 messages +func BenchmarkInsertBatch50(b *testing.B) { + if testing.Short() { + b.Skip("Skipping benchmark in short mode") + } + b.Skip("Requires database connection - run with integration tests") +} + +// BenchmarkInsertBatch100 benchmarks batch insertion with 100 messages +func BenchmarkInsertBatch100(b *testing.B) { + if testing.Short() { + b.Skip("Skipping benchmark in short mode") + } + b.Skip("Requires database connection - run with integration tests") +} + +// BenchmarkInsertRaw benchmarks raw message insertion +func BenchmarkInsertRaw(b *testing.B) { + if testing.Short() { + b.Skip("Skipping benchmark in short mode") + } + b.Skip("Requires database connection - run with integration tests") +} + +// BenchmarkMapperToDBRow benchmarks the mapping from DTO to DB row +// This doesn't require a database connection +func BenchmarkMapperToDBRow(b *testing.B) { + m := mapper.NewTelegramMapper() + telegram := createBenchmarkTelegram() + + b.ResetTimer() + b.ReportAllocs() + + for i := 0; i < b.N; i++ { + _, _ = m.ToDBRow(telegram) + } +} + +// BenchmarkMapperToDBRowBatch benchmarks mapping multiple telegrams +func BenchmarkMapperToDBRowBatch(b *testing.B) { + m := mapper.NewTelegramMapper() + telegrams := createBenchmarkTelegrams(100) + + b.ResetTimer() + b.ReportAllocs() + + for i := 0; i < b.N; i++ { + for _, tg := range telegrams { + _, _ = m.ToDBRow(tg) + } + } +} diff --git a/pkg/di/wire.go b/pkg/di/wire.go index f1b2c42..6f42bbf 100644 --- a/pkg/di/wire.go +++ b/pkg/di/wire.go @@ -1,5 +1,4 @@ //go:build wireinject -// +build wireinject package di