diff --git a/.coderabbit.yml b/.coderabbit.yml new file mode 100644 index 0000000..ed3c6a2 --- /dev/null +++ b/.coderabbit.yml @@ -0,0 +1,63 @@ +version: 1 + +reviews: + auto_review: + enabled: true + + auto_comment: + enabled: true + + auto_approve: + enabled: false + + profiles: + - profile: correctness + enabled: true + - profile: maintainability + enabled: true + - profile: security + enabled: true + - profile: performance + enabled: true + + tools: + golangci-lint: + enabled: true + +global: + language: "zh" + tone_instructions: > + 请以专业、简洁、工程化方式进行审查。 + 特别关注 Go 并发安全、context 使用、错误处理规范、资源泄漏。 + 对 NATS JetStream 消费者、pgx 连接池、数据库事务、Echo 处理器的实现进行严格检查。 + 审查 Clean Architecture 层之间的依赖是否正确。 + +paths: + include: + - "cmd/**" + - "internal/**" + - "pkg/**" + + exclude: + - "vendor/**" + - "**/*.generated.go" + - "**/mocks/**" + - "scripts/**" + - "deploy/**" + - "docs/**" + - "public/**" + - "assets/**" + - "web/**" + - "examples/**" + - "testdata/**" + +pull_requests: + disable_review_on_draft: true + +summaries: + enabled: true + format: markdown + +commit_messages: + guidance: true + enhanced: true diff --git a/.gitignore b/.gitignore index 20d4020..d8fb1c5 100644 --- a/.gitignore +++ b/.gitignore @@ -49,4 +49,14 @@ go.sum # test report *.report -coverage.* \ No newline at end of file +coverage.* +coverage.html +coverprofile.out + +# Generated files +*_gen.go +*.pb.go +pkg/di/wire_gen.go + +# Build artifacts +main \ No newline at end of file diff --git a/Makefile b/Makefile index f963d94..04449cc 100644 --- a/Makefile +++ b/Makefile @@ -68,6 +68,16 @@ fmt: ## Format Go code @echo "Formatting code..." @go fmt ./... +.PHONY: wire +wire: ## Generate wire dependency injection code + @command -v wire >/dev/null || (echo "Please install wire (go install github.com/google/wire/cmd/wire@latest)"; exit 1) + @echo "Generating wire code..." + @wire ./pkg/di + +.PHONY: generate +generate: wire ## Generate all code (wire, etc.) + @echo "Code generation complete" + .PHONY: deps deps: ## Sync go.mod / go.sum @echo "Tidying go modules..." diff --git a/README.md b/README.md index eb7b0e2..4aa2696 100644 --- a/README.md +++ b/README.md @@ -11,20 +11,25 @@ This project follows Clean Architecture principles with clear separation of conc ``` /cmd/main/main.go # Application entry point /internal - /app # Application layer (business logic orchestration) - processor.go # Message processor - /domain # Domain models (pure Go types, no external dependencies) - aviation.go # ParsedMessage and related types - /adapter # Adapter layer (interfaces and implementations) - /parser # Message parsing adapters - /mapper # Data mapping (domain ↔ infrastructure) + /port # Port layer (interfaces/contracts) repository.go # Repository interface publisher.go # Publisher interface + /domain # Domain models (pure Go types, no external dependencies) + aviation.go # Aviation domain types (FPL, DEP, ARR, etc.) + /app # Application layer (business logic orchestration) + processor.go # Message processor + /adapter # Adapter layer (implementations) + /parser # Message parsing adapters + /mapper # Data mapping (domain ↔ infrastructure) + /dto # Data Transfer Objects + telegram.go # ParsedTelegram and MessageStatus /infra # Infrastructure layer /config # Configuration management (Koanf) /nats # NATS JetStream client /postgres # PostgreSQL repository (pgx) /log # Logging (Zap) + /metrics # Prometheus metrics + /telemetry # OpenTelemetry tracing /pkg/di # Dependency injection (Wire) ``` @@ -59,7 +64,7 @@ go mod download 3. Set up PostgreSQL database: ```bash -psql -U postgres -f internal/repository/telegrams.ddl +psql -U postgres -f internal/infra/postgres/telegrams.ddl ``` 4. Configure the application: @@ -246,7 +251,7 @@ The processor exposes three complementary observability surfaces: - Application code records these via a thin `telemetry.Recorder` abstraction, which fans out to OTEL and Prometheus backends as configured. 2. **Prometheus metrics (`/metrics`)** - - Implemented in `internal/observability/metrics` and considered the primary source for SRE PromQL/SLOs. + - Implemented in `internal/infra/metrics` and considered the primary source for SRE PromQL/SLOs. - Key metric families: - `caatsm_messages_total{stream,consumer,result}` – per-stream/consumer throughput and results. - `caatsm_handle_latency_seconds_bucket{stream,consumer}` – end-to-end handling latency from NATS receive to handler completion. @@ -541,7 +546,7 @@ Flight plan messages contain detailed flight planning information. 生成的电报遵循与解析器相同的格式约定： - ARR / DEP：支持无 SSR、合法简单 SSR，以及刻意构造为“当前正则无法解析”的复杂 SSR。 -- CNL / DLA：与 `internal/parsers/aviation_parser_test.go` 中的测试样例同一类结构。 +- CNL / DLA：与 `internal/adapter/parser/aviation_parser_test.go` 中的测试样例同一类结构。 - FPL：生成包含多行 route 与 `OtherInfo` 字段的完整 FPL，`OtherInfo` 中会随机组合 `PBN/`, `NAV/`, `REG/`, `EET/`, `SEL/`, `PER/`, `RIF/`, `RMK/` 等片段，以覆盖解析逻辑。 ### 命令行参数 @@ -697,7 +702,7 @@ type ParsedMessage struct { ## Database Schema -The application uses the `aviation.telegrams` table. See `internal/repository/telegrams.ddl` for the schema definition. +The application uses the `aviation.telegrams` table. See `internal/infra/postgres/telegrams.ddl` for the schema definition. Key fields: - `uuid`: Primary key (UUID) diff --git a/Taskfile.yml b/Taskfile.yml index f5e6342..c5913f5 100644 --- a/Taskfile.yml +++ b/Taskfile.yml @@ -108,6 +108,23 @@ tasks: - echo "Linting code..." - golangci-lint run ./... + wire: + desc: Generate wire dependency injection code + cmds: + - | + if ! command -v wire >/dev/null; then + echo "Install wire: go install github.com/google/wire/cmd/wire@latest" + exit 1 + fi + - echo "Generating wire code..." + - wire ./pkg/di + + generate: + desc: Generate all code (wire, etc.) + cmds: + - task: wire + - echo "Code generation complete" + deps: desc: Sync go.mod / go.sum cmds: diff --git a/cmd/seed-telegrams/main.go b/cmd/seed-telegrams/main.go index bcc973f..a6bdbb4 100644 --- a/cmd/seed-telegrams/main.go +++ b/cmd/seed-telegrams/main.go @@ -260,6 +260,10 @@ func sendTelegram(iteration int, cfg SeedConfig, categories []string, statuses [ blob, _ := json.MarshalIndent(payload, "", " ") fmt.Println(string(blob)) fmt.Println("---") + // Still call publisher in DryRun mode for test purposes + if publisher != nil { + return publisher(payload) + } return nil } diff --git a/docker-compose.dev.yml b/docker-compose.dev.yml index 835b9d9..81f5585 100644 --- a/docker-compose.dev.yml +++ b/docker-compose.dev.yml @@ -10,7 +10,7 @@ services: - "5432:5432" volumes: - postgres-data:/var/lib/postgresql/data - - ./internal/repository/telegrams.ddl:/docker-entrypoint-initdb.d/01-telegrams.sql:ro + - ./internal/infra/postgres/telegrams.ddl:/docker-entrypoint-initdb.d/01-telegrams.sql:ro healthcheck: test: ["CMD-SHELL", "pg_isready -U $$POSTGRES_USER -d $$POSTGRES_DB"] interval: 5s @@ -31,7 +31,7 @@ services: psql --host=postgres --username=caatsm --dbname=aviation --file=/tmp/telegrams.ddl " volumes: - - ./internal/repository/telegrams.ddl:/tmp/telegrams.ddl:ro + - ./internal/infra/postgres/telegrams.ddl:/tmp/telegrams.ddl:ro restart: "no" networks: - devnet diff --git a/docs/architecture-ha.md b/docs/architecture-ha.md index 56e4f59..13616d4 100644 --- a/docs/architecture-ha.md +++ b/docs/architecture-ha.md @@ -12,7 +12,7 @@ - **Receiver service (`caatsm`)**: - NATS JetStream **pull consumer** (`internal/infra/nats/consumer.go`). - - Telegram parser and domain model (`internal/app`, `internal/parsers`, `internal/domain`). + - Telegram parser and domain model (`internal/app`, `internal/adapter/parser`, `internal/domain`). - PostgreSQL repository (`internal/infra/postgres`). - Monitoring/observability server (`internal/infra/monitoring`). diff --git a/docs/dev-guide.md b/docs/dev-guide.md index 49a1b51..1001a89 100644 --- a/docs/dev-guide.md +++ b/docs/dev-guide.md @@ -12,7 +12,7 @@ docker compose -f docker-compose.dev.yml up -d postgres nats nats-box > Development mode defaults to `nats.mode = "core"`, so the processor consumes directly from the configured subject (`subscription.topic`). **However, the publisher always targets JetStream for deduplicated fan-out, so the provided Taskfile (and most examples below) override the mode to `jetstream`.** If you truly need core mode, set `CAATSM_NATS_MODE=core` manually and ensure any publishers use core subjects. -- `postgres` seeds the `aviation` schema using `internal/repository/telegrams.ddl` and exposes port `5432`. +- `postgres` seeds the `aviation` schema using `internal/infra/postgres/telegrams.ddl` and exposes port `5432`. - `nats` enables JetStream with client port `4222` and monitoring/UI on `8222`. - `nats-box` provides a toolbox container (`docker compose exec nats-box sh`) for publishing test messages or inspecting JetStream. - `nats-exporter` scrapes the monitoring endpoints (`/varz`, `/connz`, `/routez`, `/subz`) and exposes them as Prometheus metrics on port `7777` for the Grafana dashboards. @@ -62,7 +62,7 @@ Use these tasks if you prefer a one-command workflow instead of invoking `docker ## Publishing Sample Telegrams -Use the helper CLI in `cmd/seed-telegrams` to push realistic payloads onto NATS (mirrors the fixtures in `internal/parsers/aviation_parser_test.go`): +Use the helper CLI in `cmd/seed-telegrams` to push realistic payloads onto NATS (mirrors the fixtures in `internal/adapter/parser/aviation_parser_test.go`): ```bash # Insert rows into aviation.telegrams_raw and publish to NATS simultaneously @@ -149,7 +149,7 @@ Services: ## Troubleshooting -- **PostgreSQL init errors**: ensure `internal/repository/telegrams.ddl` is valid SQL and the `postgres-data` volume is removed (`docker volume rm go-caatsm_postgres-data`) before restarting. +- **PostgreSQL init errors**: ensure `internal/infra/postgres/telegrams.ddl` is valid SQL and the `postgres-data` volume is removed (`docker volume rm go-caatsm_postgres-data`) before restarting. - **NATS connection failures**: confirm ports `4222/8222` are free and JetStream is enabled; use `docker compose logs nats`. - **Prometheus scrape failures**: verify endpoints listed in `configs/prometheus.dev.yml` match the service names defined in Docker Compose. - **Grafana provisioning issues**: check container logs (`docker compose logs grafana`) to ensure the datasources file was read; correct file permissions or YAML formatting if provisioning is skipped. diff --git a/docs/observability.md b/docs/observability.md index 403c3ff..f70bb55 100644 --- a/docs/observability.md +++ b/docs/observability.md @@ -177,7 +177,7 @@ Tracing is configured via the `telemetry` section: The receiver reports two complementary sets of metrics: - **Prometheus metrics via `/metrics`** - Implemented in `internal/observability/metrics`, covering: + Implemented in `internal/infra/metrics`, covering: - End-to-end message handling (`caatsm_messages_total`, `caatsm_handle_latency_seconds`, `caatsm_retries_total`) - DB activity (`caatsm_db_queries_total`, @@ -223,7 +223,7 @@ Important attributes: ### Structured Logging Contract -Logging is done with Zap. The `internal/observability/logging` package standardises fields via `MessageFields`: +Logging is done with Zap. The `internal/infra/log` package standardises fields via `MessageFields`: - `service` – logical component (`caatsm-consumer`, `caatsm-processor` etc.). - `transport_msg_id` – NATS/envelope message ID (derived from `Nats-Msg-Id` or JetStream sequence). diff --git a/go.mod b/go.mod index b6170ea..c99cd55 100644 --- a/go.mod +++ b/go.mod @@ -13,7 +13,7 @@ require ( github.com/nats-io/nats.go v1.47.0 github.com/onsi/ginkgo/v2 v2.27.2 github.com/onsi/gomega v1.38.2 - github.com/prometheus/client_golang v1.20.3 + github.com/prometheus/client_golang v1.23.2 github.com/testcontainers/testcontainers-go v0.30.0 github.com/urfave/cli/v2 v2.27.7 go.opentelemetry.io/otel v1.38.0 @@ -77,9 +77,9 @@ require ( github.com/pelletier/go-toml v1.9.5 // indirect github.com/pkg/errors v0.9.1 // indirect github.com/power-devops/perfstat v0.0.0-20210106213030-5aafc221ea8c // indirect - github.com/prometheus/client_model v0.6.1 // indirect - github.com/prometheus/common v0.55.0 // indirect - github.com/prometheus/procfs v0.15.1 // indirect + github.com/prometheus/client_model v0.6.2 // indirect + github.com/prometheus/common v0.67.2 // indirect + github.com/prometheus/procfs v0.19.2 // indirect github.com/russross/blackfriday/v2 v2.1.0 // indirect github.com/shirou/gopsutil/v3 v3.23.12 // indirect github.com/shoenig/go-m1cpu v0.1.6 // indirect @@ -93,6 +93,7 @@ require ( go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.38.0 // indirect go.opentelemetry.io/proto/otlp v1.9.0 // indirect go.uber.org/multierr v1.11.0 // indirect + go.yaml.in/yaml/v2 v2.4.3 // indirect go.yaml.in/yaml/v3 v3.0.4 // indirect golang.org/x/crypto v0.44.0 // indirect golang.org/x/exp v0.0.0-20230510235704-dd950f8aeaea // indirect diff --git a/internal/model/telegram.go b/internal/adapter/dto/telegram.go similarity index 99% rename from internal/model/telegram.go rename to internal/adapter/dto/telegram.go index 510aee3..59eac43 100644 --- a/internal/model/telegram.go +++ b/internal/adapter/dto/telegram.go @@ -1,4 +1,4 @@ -package model +package dto import ( "time" diff --git a/internal/adapter/mapper/mapper.go b/internal/adapter/mapper/mapper.go index 81505f7..8ec83cd 100644 --- a/internal/adapter/mapper/mapper.go +++ b/internal/adapter/mapper/mapper.go @@ -1,13 +1,13 @@ package mapper -import "caatsm/internal/model" +import "caatsm/internal/adapter/dto" // Mapper defines the interface for mapping between pipeline models and database models type Mapper interface { // ToDBRow converts a ParsedTelegram to a database row representation - ToDBRow(msg *model.ParsedTelegram) ([]interface{}, error) + ToDBRow(msg *dto.ParsedTelegram) ([]interface{}, error) // FromDBRow converts a database row to a ParsedTelegram - FromDBRow(row []interface{}) (*model.ParsedTelegram, error) + FromDBRow(row []interface{}) (*dto.ParsedTelegram, error) } diff --git a/internal/adapter/mapper/telegram.go b/internal/adapter/mapper/telegram.go index b093fbc..dd35820 100644 --- a/internal/adapter/mapper/telegram.go +++ b/internal/adapter/mapper/telegram.go @@ -1,7 +1,7 @@ package mapper import ( - "caatsm/internal/model" + "caatsm/internal/adapter/dto" "encoding/json" "fmt" "time" @@ -18,7 +18,7 @@ func NewTelegramMapper() *TelegramMapper { } // ToDBRow converts a ParsedTelegram to a database row representation -func (m *TelegramMapper) ToDBRow(msg *model.ParsedTelegram) ([]interface{}, error) { +func (m *TelegramMapper) ToDBRow(msg *dto.ParsedTelegram) ([]interface{}, error) { // Parse UUID var msgUUID uuid.UUID var err error @@ -45,7 +45,7 @@ func (m *TelegramMapper) ToDBRow(msg *model.ParsedTelegram) ([]interface{}, erro status := msg.Status if status == "" { - status = model.MessageStatusUnknown + status = dto.MessageStatusUnknown } return []interface{}{ @@ -70,7 +70,7 @@ func (m *TelegramMapper) ToDBRow(msg *model.ParsedTelegram) ([]interface{}, erro } // FromDBRow converts a database row to a ParsedTelegram -func (m *TelegramMapper) FromDBRow(row []interface{}) (*model.ParsedTelegram, error) { +func (m *TelegramMapper) FromDBRow(row []interface{}) (*dto.ParsedTelegram, error) { const expectedColumns = 17 if len(row) < expectedColumns { return nil, fmt.Errorf("expected %d columns, got %d", expectedColumns, len(row)) @@ -128,12 +128,12 @@ func (m *TelegramMapper) FromDBRow(row []interface{}) (*model.ParsedTelegram, er } } - status := model.MessageStatusUnknown + status := dto.MessageStatusUnknown if rawStatus := toString(row[11]); rawStatus != "" { - status = model.MessageStatus(rawStatus) + status = dto.MessageStatus(rawStatus) } - return &model.ParsedTelegram{ + return &dto.ParsedTelegram{ Uuid: msgUUID.String(), MessageID: toString(row[1]), DateTime: toString(row[2]), diff --git a/internal/adapter/mapper/telegram_test.go b/internal/adapter/mapper/telegram_test.go index b7afe43..917581d 100644 --- a/internal/adapter/mapper/telegram_test.go +++ b/internal/adapter/mapper/telegram_test.go @@ -3,7 +3,7 @@ package mapper import ( "time" - "caatsm/internal/model" + "caatsm/internal/adapter/dto" "github.com/google/uuid" . "github.com/onsi/ginkgo/v2" @@ -19,7 +19,7 @@ var _ = Describe("TelegramMapper", func() { Describe("ToDBRow", func() { It("generates a UUID when missing", func() { - msg := &model.ParsedTelegram{} + msg := &dto.ParsedTelegram{} row, err := mapper.ToDBRow(msg) Expect(err).NotTo(HaveOccurred()) @@ -33,7 +33,7 @@ var _ = Describe("TelegramMapper", func() { Describe("FromDBRow", func() { It("round-trips telegram data", func() { now := time.Now().UTC() - original := &model.ParsedTelegram{ + original := &dto.ParsedTelegram{ Uuid: uuid.NewString(), MessageID: "TMQ1324", DateTime: "150631", @@ -49,7 +49,7 @@ var _ = Describe("TelegramMapper", func() { ParsedAt: now, DispatchedAt: now, NeedDispatch: true, - Status: model.MessageStatusParsed, + Status: dto.MessageStatusParsed, } row, err := mapper.ToDBRow(original) diff --git a/internal/parsers/aviation_parser.go b/internal/adapter/parser/aviation.go similarity index 97% rename from internal/parsers/aviation_parser.go rename to internal/adapter/parser/aviation.go index a815809..7cb8a83 100644 --- a/internal/parsers/aviation_parser.go +++ b/internal/adapter/parser/aviation.go @@ -1,8 +1,8 @@ -package parsers +package parser import ( "caatsm/internal/domain" - "caatsm/internal/model" + "caatsm/internal/adapter/dto" "errors" "fmt" "regexp" @@ -200,14 +200,14 @@ func (parser *BodyParser) createBodyData(data map[string]string) (string, interf } } -func Parse(rawText string) (*model.ParsedTelegram, error) { +func Parse(rawText string) (*dto.ParsedTelegram, error) { header, err := ParseHeader(rawText) if err != nil { - msg := model.NewParsedTelegram() + msg := dto.NewParsedTelegram() msg.Content = rawText msg.Comments = err.Error() msg.ErrorReason = err.Error() - msg.Status = model.MessageStatusHeaderError + msg.Status = dto.MessageStatusHeaderError return msg, fmt.Errorf("%w: %w", ErrHeaderParse, err) } @@ -217,7 +217,7 @@ func Parse(rawText string) (*model.ParsedTelegram, error) { header.ParsedAt = time.Now() if bodyErr != nil { - return &model.ParsedTelegram{ + return &dto.ParsedTelegram{ MessageID: header.MessageID, DateTime: header.DateTime, PriorityIndicator: header.PriorityIndicator, @@ -232,12 +232,12 @@ func Parse(rawText string) (*model.ParsedTelegram, error) { ParsedAt: header.ParsedAt, Parsed: false, Comments: bodyErr.Error(), - Status: model.MessageStatusBodyError, + Status: dto.MessageStatusBodyError, ErrorReason: bodyErr.Error(), }, fmt.Errorf("%w: %w", ErrBodyParse, bodyErr) } - parsed := &model.ParsedTelegram{ + parsed := &dto.ParsedTelegram{ MessageID: header.MessageID, DateTime: header.DateTime, PriorityIndicator: header.PriorityIndicator, @@ -252,7 +252,7 @@ func Parse(rawText string) (*model.ParsedTelegram, error) { ReceivedAt: header.ReceivedAt, ParsedAt: header.ParsedAt, Parsed: true, - Status: model.MessageStatusParsed, + Status: dto.MessageStatusParsed, ErrorReason: "", } diff --git a/internal/adapter/parser/aviation_parser.go b/internal/adapter/parser/aviation_parser.go deleted file mode 100644 index 000c1f1..0000000 --- a/internal/adapter/parser/aviation_parser.go +++ /dev/null @@ -1,20 +0,0 @@ -package parser - -import ( - "caatsm/internal/model" - "caatsm/internal/parsers" -) - -// AviationParser implements the Parser interface using the existing parsers package -type AviationParser struct{} - -// NewAviationParser creates a new aviation parser -func NewAviationParser() *AviationParser { - return &AviationParser{} -} - -// Parse parses a raw message string and returns a ParsedTelegram -func (p *AviationParser) Parse(rawText string) (*model.ParsedTelegram, error) { - // Use the existing Parse function from internal/parsers - return parsers.Parse(rawText) -} diff --git a/internal/parsers/aviation_parser_test.go b/internal/adapter/parser/aviation_parser_test.go similarity index 99% rename from internal/parsers/aviation_parser_test.go rename to internal/adapter/parser/aviation_parser_test.go index 38638d8..5d5e517 100644 --- a/internal/parsers/aviation_parser_test.go +++ b/internal/adapter/parser/aviation_parser_test.go @@ -1,4 +1,4 @@ -package parsers +package parser import ( "caatsm/internal/domain" diff --git a/internal/parsers/constants.go b/internal/adapter/parser/constants.go similarity index 99% rename from internal/parsers/constants.go rename to internal/adapter/parser/constants.go index 86e18eb..6102520 100644 --- a/internal/parsers/constants.go +++ b/internal/adapter/parser/constants.go @@ -1,4 +1,4 @@ -package parsers +package parser import "regexp" diff --git a/internal/adapter/parser/parser.go b/internal/adapter/parser/parser.go index 08ecfb2..d12435c 100644 --- a/internal/adapter/parser/parser.go +++ b/internal/adapter/parser/parser.go @@ -1,9 +1,9 @@ package parser -import "caatsm/internal/model" +import "caatsm/internal/adapter/dto" // Parser defines the interface for parsing raw telegram messages type Parser interface { // Parse parses a raw message string and returns a ParsedTelegram - Parse(rawText string) (*model.ParsedTelegram, error) + Parse(rawText string) (*dto.ParsedTelegram, error) } diff --git a/internal/parsers/pattern.go b/internal/adapter/parser/pattern.go similarity index 99% rename from internal/parsers/pattern.go rename to internal/adapter/parser/pattern.go index d132278..028de80 100644 --- a/internal/parsers/pattern.go +++ b/internal/adapter/parser/pattern.go @@ -1,4 +1,4 @@ -package parsers +package parser import ( "regexp" diff --git a/internal/parsers/pattern_test.go b/internal/adapter/parser/pattern_test.go similarity index 98% rename from internal/parsers/pattern_test.go rename to internal/adapter/parser/pattern_test.go index 553d3ad..e52b9bb 100644 --- a/internal/parsers/pattern_test.go +++ b/internal/adapter/parser/pattern_test.go @@ -1,4 +1,4 @@ -package parsers +package parser import ( . "github.com/onsi/ginkgo/v2" diff --git a/internal/adapter/parser/provider.go b/internal/adapter/parser/provider.go index 24f0ad1..d4664f3 100644 --- a/internal/adapter/parser/provider.go +++ b/internal/adapter/parser/provider.go @@ -1,7 +1,17 @@ package parser +import "caatsm/internal/adapter/dto" + +// AviationParser implements the Parser interface +type AviationParser struct{} + +// Parse parses a raw message string and returns a ParsedTelegram +func (p *AviationParser) Parse(rawText string) (*dto.ParsedTelegram, error) { + return Parse(rawText) +} + // ProvideParser creates a parser instance func ProvideParser() Parser { - return NewAviationParser() + return &AviationParser{} } diff --git a/internal/parsers/schedule_parser.go b/internal/adapter/parser/schedule.go similarity index 99% rename from internal/parsers/schedule_parser.go rename to internal/adapter/parser/schedule.go index f479d7d..53a8d30 100644 --- a/internal/parsers/schedule_parser.go +++ b/internal/adapter/parser/schedule.go @@ -1,4 +1,4 @@ -package parsers +package parser import ( "caatsm/internal/domain" diff --git a/internal/parsers/schedule_parser_test.go b/internal/adapter/parser/schedule_parser_test.go similarity index 99% rename from internal/parsers/schedule_parser_test.go rename to internal/adapter/parser/schedule_parser_test.go index f8d9e45..582b295 100644 --- a/internal/parsers/schedule_parser_test.go +++ b/internal/adapter/parser/schedule_parser_test.go @@ -1,4 +1,4 @@ -package parsers +package parser import ( "strings" diff --git a/internal/parsers/suite_test.go b/internal/adapter/parser/suite_test.go similarity index 91% rename from internal/parsers/suite_test.go rename to internal/adapter/parser/suite_test.go index 6bceaff..740e031 100644 --- a/internal/parsers/suite_test.go +++ b/internal/adapter/parser/suite_test.go @@ -1,4 +1,4 @@ -package parsers +package parser import ( "testing" diff --git a/internal/app/processor.go b/internal/app/processor.go index 60da066..70bd01b 100644 --- a/internal/app/processor.go +++ b/internal/app/processor.go @@ -1,11 +1,11 @@ package app import ( - "caatsm/internal/adapter" "caatsm/internal/adapter/parser" - "caatsm/internal/model" - obslogging "caatsm/internal/observability/logging" - "caatsm/internal/observability/telemetry" + "caatsm/internal/adapter/dto" + "caatsm/internal/infra/log" + "caatsm/internal/infra/telemetry" + "caatsm/internal/port" "context" "fmt" "strings" @@ -21,15 +21,15 @@ import ( // MessageProcessor handles message processing type MessageProcessor struct { parser parser.Parser - repository adapter.Repository - publisher adapter.Publisher + repository port.Repository + publisher port.Publisher logger *zap.Logger telemetry telemetry.Recorder } // ProcessingStatus represents the outcome of the processing pipeline // (persistence, publishing, etc.), independent from the parsing status -// captured in model.MessageStatus. +// captured in dto.MessageStatus. type ProcessingStatus string const ( @@ -41,8 +41,8 @@ const ( // NewMessageProcessor creates a new message processor func NewMessageProcessor( parser parser.Parser, - repository adapter.Repository, - publisher adapter.Publisher, + repository port.Repository, + publisher port.Publisher, rec telemetry.Recorder, logger *zap.Logger, ) *MessageProcessor { @@ -70,10 +70,10 @@ func (p *MessageProcessor) Handle(ctx context.Context, raw []byte, msgID string) parsed, parseErr := p.parser.Parse(string(raw)) if parsed == nil { - parsed = model.NewParsedTelegram() + parsed = dto.NewParsedTelegram() parsed.Content = string(raw) parsed.ErrorReason = "parser returned nil" - parsed.Status = model.MessageStatusBodyError + parsed.Status = dto.MessageStatusBodyError parseErr = fmt.Errorf("parser returned nil") } @@ -90,11 +90,11 @@ func (p *MessageProcessor) Handle(ctx context.Context, raw []byte, msgID string) if parsed.ParsedAt.IsZero() { parsed.ParsedAt = time.Now() } - if parsed.Status == model.MessageStatusUnknown { + if parsed.Status == dto.MessageStatusUnknown { if parseErr == nil { - parsed.Status = model.MessageStatusParsed + parsed.Status = dto.MessageStatusParsed } else { - parsed.Status = model.MessageStatusBodyError + parsed.Status = dto.MessageStatusBodyError } } @@ -105,7 +105,7 @@ func (p *MessageProcessor) Handle(ctx context.Context, raw []byte, msgID string) attribute.String("telegram.status", string(parsed.Status)), ) - msgLogger := obslogging.WithMessageContext(p.logger, obslogging.MessageFields{ + msgLogger := log.WithMessageContext(p.logger, log.MessageFields{ Service: "caatsm-processor", TransportMsgID: msgID, BusinessMsgID: parsed.MessageID, @@ -191,7 +191,7 @@ func (p *MessageProcessor) Handle(ctx context.Context, raw []byte, msgID string) return nil } -func (p *MessageProcessor) persistRaw(ctx context.Context, msg *model.ParsedTelegram) { +func (p *MessageProcessor) persistRaw(ctx context.Context, msg *dto.ParsedTelegram) { if msg == nil || p.repository == nil { return } diff --git a/internal/app/processor_test.go b/internal/app/processor_test.go index a49936e..a9de2ef 100644 --- a/internal/app/processor_test.go +++ b/internal/app/processor_test.go @@ -6,10 +6,10 @@ import ( "strings" "time" - "caatsm/internal/adapter" "caatsm/internal/adapter/parser" - "caatsm/internal/model" - "caatsm/internal/observability/telemetry" + "caatsm/internal/adapter/dto" + "caatsm/internal/infra/telemetry" + "caatsm/internal/port" "github.com/google/uuid" . "github.com/onsi/ginkgo/v2" @@ -52,7 +52,7 @@ var _ = Describe("MessageProcessor", func() { It("preserves UUIDs and appends nats message id comment", func() { originalUUID := uuid.NewString() - parserStub.value = &model.ParsedTelegram{Uuid: originalUUID, Parsed: true, Status: model.MessageStatusParsed} + parserStub.value = &dto.ParsedTelegram{Uuid: originalUUID, Parsed: true, Status: dto.MessageStatusParsed} Expect(proc.Handle(ctx, []byte("payload"), "msg-123")).To(Succeed()) @@ -63,7 +63,7 @@ var _ = Describe("MessageProcessor", func() { }) It("treats publisher failures as permanent and stores raw entries", func() { - parserStub.value = &model.ParsedTelegram{Parsed: true, Status: model.MessageStatusParsed} + parserStub.value = &dto.ParsedTelegram{Parsed: true, Status: dto.MessageStatusParsed} pub.err = errors.New("publish failed") err := proc.Handle(ctx, []byte("payload"), "id-3") @@ -71,15 +71,15 @@ var _ = Describe("MessageProcessor", func() { Expect(IsPermanent(err)).To(BeTrue()) Expect(repo.last()).NotTo(BeNil()) Expect(repo.rawCount()).To(Equal(1)) - Expect(repo.lastRaw().Status).To(Equal(model.MessageStatusParsed)) + Expect(repo.lastRaw().Status).To(Equal(dto.MessageStatusParsed)) Expect(repo.lastRaw().ErrorReason).To(ContainSubstring("publish failed")) }) It("sets timestamps when missing", func() { - parserStub.value = &model.ParsedTelegram{ + parserStub.value = &dto.ParsedTelegram{ Uuid: uuid.NewString(), Parsed: true, - Status: model.MessageStatusParsed, + Status: dto.MessageStatusParsed, } pub.err = nil @@ -97,10 +97,10 @@ var _ = Describe("MessageProcessor", func() { It("does not override provided timestamps", func() { received := time.Now().Add(-2 * time.Minute) parsedAt := time.Now().Add(-1 * time.Minute) - parserStub.value = &model.ParsedTelegram{ + parserStub.value = &dto.ParsedTelegram{ Uuid: uuid.NewString(), Parsed: true, - Status: model.MessageStatusParsed, + Status: dto.MessageStatusParsed, ReceivedAt: received, ParsedAt: parsedAt, } @@ -114,10 +114,10 @@ var _ = Describe("MessageProcessor", func() { core, logs := observer.New(zap.WarnLevel) logger := zap.New(core) parserStub = &stubParser{ - value: &model.ParsedTelegram{ + value: &dto.ParsedTelegram{ Content: strings.Repeat("x", 1024), Parsed: false, - Status: model.MessageStatusBodyError, + Status: dto.MessageStatusBodyError, ErrorReason: "parse failure", }, err: errors.New("parse failure"), @@ -145,27 +145,27 @@ var _ = Describe("MessageProcessor", func() { }) }) -func newTestProcessor(p parser.Parser, repo adapter.Repository, pub adapter.Publisher) *MessageProcessor { +func newTestProcessor(p parser.Parser, repo port.Repository, pub port.Publisher) *MessageProcessor { return NewMessageProcessor(p, repo, pub, telemetry.NewNoop(), zap.NewNop()) } type stubParser struct { - value *model.ParsedTelegram + value *dto.ParsedTelegram err error } -func (s *stubParser) Parse(rawText string) (*model.ParsedTelegram, error) { +func (s *stubParser) Parse(rawText string) (*dto.ParsedTelegram, error) { return s.value, s.err } type stubRepository struct { - inserted []*model.ParsedTelegram - raw []*model.ParsedTelegram + inserted []*dto.ParsedTelegram + raw []*dto.ParsedTelegram err error rawErr error } -func (s *stubRepository) InsertOne(ctx context.Context, msg *model.ParsedTelegram) error { +func (s *stubRepository) InsertOne(ctx context.Context, msg *dto.ParsedTelegram) error { if s.err != nil { return s.err } @@ -173,11 +173,11 @@ func (s *stubRepository) InsertOne(ctx context.Context, msg *model.ParsedTelegra return nil } -func (s *stubRepository) InsertBatch(ctx context.Context, msgs []*model.ParsedTelegram) error { +func (s *stubRepository) InsertBatch(ctx context.Context, msgs []*dto.ParsedTelegram) error { return errors.New("not implemented") } -func (s *stubRepository) InsertRaw(ctx context.Context, msg *model.ParsedTelegram) error { +func (s *stubRepository) InsertRaw(ctx context.Context, msg *dto.ParsedTelegram) error { if s.rawErr != nil { return s.rawErr } @@ -185,14 +185,14 @@ func (s *stubRepository) InsertRaw(ctx context.Context, msg *model.ParsedTelegra return nil } -func (s *stubRepository) last() *model.ParsedTelegram { +func (s *stubRepository) last() *dto.ParsedTelegram { if len(s.inserted) == 0 { return nil } return s.inserted[len(s.inserted)-1] } -func (s *stubRepository) lastRaw() *model.ParsedTelegram { +func (s *stubRepository) lastRaw() *dto.ParsedTelegram { if len(s.raw) == 0 { return nil } diff --git a/internal/domain/aviation.go b/internal/domain/aviation.go index 4c556b0..ae74dce 100644 --- a/internal/domain/aviation.go +++ b/internal/domain/aviation.go @@ -83,5 +83,5 @@ NeedDispatch: false. */ // NOTE: Parsed telegram pipeline structures (ParsedTelegram, MessageStatus, etc.) -// have been moved to the internal/model package to keep the domain layer focused +// have been moved to the internal/adapter/dto package to keep the domain layer focused // purely on aviation business concepts (FPL, DEP, ARR, etc.). diff --git a/internal/iface/interface.go b/internal/iface/interface.go deleted file mode 100644 index 54a9dcc..0000000 --- a/internal/iface/interface.go +++ /dev/null @@ -1,19 +0,0 @@ -package iface - -import "caatsm/internal/infra/config" - -type MessageHandler interface { - HandleMessage(msg []byte, id string) error -} - -type MessagePublisher interface { - Publish(message interface{}) error -} - -type MessageSubscriber interface { - Subscribe(config *config.Config) error -} - -type MessageRepository interface { - CreateNew(message interface{}) error -} diff --git a/internal/observability/logging/logger.go b/internal/infra/log/logging.go similarity index 99% rename from internal/observability/logging/logger.go rename to internal/infra/log/logging.go index 8ed0eb3..3e654bc 100644 --- a/internal/observability/logging/logger.go +++ b/internal/infra/log/logging.go @@ -1,4 +1,4 @@ -package logging +package log import "go.uber.org/zap" diff --git a/internal/observability/metrics/metrics.go b/internal/infra/metrics/metrics.go similarity index 100% rename from internal/observability/metrics/metrics.go rename to internal/infra/metrics/metrics.go diff --git a/internal/infra/monitoring/server.go b/internal/infra/monitoring/server.go index b9b747f..a320f69 100644 --- a/internal/infra/monitoring/server.go +++ b/internal/infra/monitoring/server.go @@ -3,7 +3,7 @@ package monitoring import ( "caatsm/internal/infra/buildinfo" "caatsm/internal/infra/config" - obsmetrics "caatsm/internal/observability/metrics" + obsmetrics "caatsm/internal/infra/metrics" "context" "encoding/json" "errors" diff --git a/internal/infra/nats/consumer.go b/internal/infra/nats/consumer.go index 396df22..18e7304 100644 --- a/internal/infra/nats/consumer.go +++ b/internal/infra/nats/consumer.go @@ -3,9 +3,9 @@ package nats import ( "caatsm/internal/app" "caatsm/internal/infra/config" - obslogging "caatsm/internal/observability/logging" - obsmetrics "caatsm/internal/observability/metrics" - "caatsm/internal/observability/telemetry" + "caatsm/internal/infra/log" + obsmetrics "caatsm/internal/infra/metrics" + "caatsm/internal/infra/telemetry" "context" "encoding/json" "errors" @@ -792,7 +792,7 @@ func (c *Consumer) processMessage(ctx context.Context, msg *nats.Msg) error { ) } - msgLogger := obslogging.WithMessageContext(c.logger, obslogging.MessageFields{ + msgLogger := log.WithMessageContext(c.logger, log.MessageFields{ Service: "caatsm-consumer", TransportMsgID: msgID, Stream: c.streamName, diff --git a/internal/infra/nats/publisher.go b/internal/infra/nats/publisher.go index 99e3955..a5dffcd 100644 --- a/internal/infra/nats/publisher.go +++ b/internal/infra/nats/publisher.go @@ -1,9 +1,9 @@ package nats import ( - "caatsm/internal/adapter" "caatsm/internal/infra/config" - "caatsm/internal/model" + "caatsm/internal/adapter/dto" + "caatsm/internal/port" "encoding/json" "errors" "fmt" @@ -25,7 +25,7 @@ func ProvidePublisher( js nats.JetStreamContext, cfg *config.Config, logger *zap.Logger, -) (adapter.Publisher, error) { +) (port.Publisher, error) { return &Publisher{ js: js, cfg: cfg, @@ -51,7 +51,7 @@ func (p *Publisher) Publish(message interface{}) error { jsMsg.Data = messageBytes switch typed := message.(type) { - case *model.ParsedTelegram: + case *dto.ParsedTelegram: if typed != nil && typed.Uuid != "" { jsMsg.Header.Set("Nats-Msg-Id", typed.Uuid) } else { diff --git a/internal/infra/postgres/repository.go b/internal/infra/postgres/repository.go index ce65f7b..45742df 100644 --- a/internal/infra/postgres/repository.go +++ b/internal/infra/postgres/repository.go @@ -1,10 +1,10 @@ package postgres import ( - "caatsm/internal/adapter" "caatsm/internal/adapter/mapper" - "caatsm/internal/model" - obsmetrics "caatsm/internal/observability/metrics" + "caatsm/internal/adapter/dto" + "caatsm/internal/port" + obsmetrics "caatsm/internal/infra/metrics" "context" "encoding/json" "fmt" @@ -19,7 +19,7 @@ import ( "go.uber.org/zap" ) -// Repository implements the adapter.Repository interface using PostgreSQL +// Repository implements the port.Repository interface using PostgreSQL type Repository struct { pool *pgxpool.Pool mapper *mapper.TelegramMapper @@ -27,7 +27,7 @@ type Repository struct { } // ProvideRepository creates a PostgreSQL repository -func ProvideRepository(pool *pgxpool.Pool, logger *zap.Logger) (adapter.Repository, error) { +func ProvideRepository(pool *pgxpool.Pool, logger *zap.Logger) (port.Repository, error) { return &Repository{ pool: pool, mapper: mapper.NewTelegramMapper(), @@ -36,7 +36,7 @@ func ProvideRepository(pool *pgxpool.Pool, logger *zap.Logger) (adapter.Reposito } // InsertOne inserts a single telegram message -func (r *Repository) InsertOne(ctx context.Context, msg *model.ParsedTelegram) error { +func (r *Repository) InsertOne(ctx context.Context, msg *dto.ParsedTelegram) error { ctx, span := otel.Tracer("caatsm/postgres").Start(ctx, "Repository.InsertOne") defer span.End() span.SetAttributes(attribute.String("db.table", "aviation.telegrams")) @@ -114,7 +114,7 @@ func (r *Repository) InsertOne(ctx context.Context, msg *model.ParsedTelegram) e } // InsertBatch inserts multiple telegram messages in a batch using CopyFrom -func (r *Repository) InsertBatch(ctx context.Context, msgs []*model.ParsedTelegram) error { +func (r *Repository) InsertBatch(ctx context.Context, msgs []*dto.ParsedTelegram) error { ctx, span := otel.Tracer("caatsm/postgres").Start(ctx, "Repository.InsertBatch") defer span.End() span.SetAttributes(attribute.String("db.table", "aviation.telegrams")) @@ -169,7 +169,7 @@ func (r *Repository) InsertBatch(ctx context.Context, msgs []*model.ParsedTelegr } // InsertRaw inserts a failed telegram into aviation.telegrams_raw for post-processing. -func (r *Repository) InsertRaw(ctx context.Context, msg *model.ParsedTelegram) error { +func (r *Repository) InsertRaw(ctx context.Context, msg *dto.ParsedTelegram) error { ctx, span := otel.Tracer("caatsm/postgres").Start(ctx, "Repository.InsertRaw") defer span.End() span.SetAttributes(attribute.String("db.table", "aviation.telegrams_raw")) diff --git a/internal/repository/telegrams.ddl b/internal/infra/postgres/telegrams.ddl similarity index 100% rename from internal/repository/telegrams.ddl rename to internal/infra/postgres/telegrams.ddl diff --git a/internal/observability/telemetry/telemetry.go b/internal/infra/telemetry/telemetry.go similarity index 99% rename from internal/observability/telemetry/telemetry.go rename to internal/infra/telemetry/telemetry.go index 9a4e1ab..9333d1e 100644 --- a/internal/observability/telemetry/telemetry.go +++ b/internal/infra/telemetry/telemetry.go @@ -2,7 +2,7 @@ package telemetry import ( "caatsm/internal/infra/config" - obsmetrics "caatsm/internal/observability/metrics" + obsmetrics "caatsm/internal/infra/metrics" "context" "time" @@ -168,7 +168,7 @@ func (c *compositeRecorder) RecordJSAPICall(operation string) { } // promRecorder delegates to the Prometheus metrics helpers in the -// internal/observability/metrics package. +// internal/infra/metrics package. type promRecorder struct{} func newPromRecorder() Recorder { diff --git a/internal/adapter/publisher.go b/internal/port/publisher.go similarity index 91% rename from internal/adapter/publisher.go rename to internal/port/publisher.go index 947151d..1e81700 100644 --- a/internal/adapter/publisher.go +++ b/internal/port/publisher.go @@ -1,4 +1,4 @@ -package adapter +package port // Publisher defines the interface for publishing parsed messages type Publisher interface { diff --git a/internal/adapter/repository.go b/internal/port/repository.go similarity index 55% rename from internal/adapter/repository.go rename to internal/port/repository.go index f04ded1..e0a59b1 100644 --- a/internal/adapter/repository.go +++ b/internal/port/repository.go @@ -1,19 +1,19 @@ -package adapter +package port import ( "context" - "caatsm/internal/model" + "caatsm/internal/adapter/dto" ) // Repository defines the interface for message persistence type Repository interface { // InsertOne inserts a single telegram message - InsertOne(ctx context.Context, msg *model.ParsedTelegram) error + InsertOne(ctx context.Context, msg *dto.ParsedTelegram) error // InsertBatch inserts multiple telegram messages in a batch - InsertBatch(ctx context.Context, msgs []*model.ParsedTelegram) error + InsertBatch(ctx context.Context, msgs []*dto.ParsedTelegram) error // InsertRaw captures an unparsed or failed telegram for later analysis. - InsertRaw(ctx context.Context, msg *model.ParsedTelegram) error + InsertRaw(ctx context.Context, msg *dto.ParsedTelegram) error } diff --git a/internal/repository/genqlient.graphql b/internal/repository/genqlient.graphql deleted file mode 100644 index 6e6a3a6..0000000 --- a/internal/repository/genqlient.graphql +++ /dev/null @@ -1,6 +0,0 @@ -mutation newMessage($object: aviation_telegrams_insert_input!) { - insert_aviation_telegrams_one(object: $object) { - message_id - uuid - } -} \ No newline at end of file diff --git a/main b/main deleted file mode 100755 index 47774dd..0000000 Binary files a/main and /dev/null differ diff --git a/pkg/di/wire.go b/pkg/di/wire.go index 6296de6..f1b2c42 100644 --- a/pkg/di/wire.go +++ b/pkg/di/wire.go @@ -11,7 +11,7 @@ import ( "caatsm/internal/infra/monitoring" "caatsm/internal/infra/nats" "caatsm/internal/infra/postgres" - "caatsm/internal/observability/telemetry" + "caatsm/internal/infra/telemetry" "github.com/google/wire" ) diff --git a/pkg/di/wire_gen.go b/pkg/di/wire_gen.go index 7828639..833bd80 100644 --- a/pkg/di/wire_gen.go +++ b/pkg/di/wire_gen.go @@ -14,7 +14,7 @@ import ( "caatsm/internal/infra/monitoring" "caatsm/internal/infra/nats" "caatsm/internal/infra/postgres" - "caatsm/internal/observability/telemetry" + "caatsm/internal/infra/telemetry" "github.com/google/wire" ) diff --git a/test/integration/jetstream_to_timescale_test.go b/test/integration/jetstream_to_timescale_test.go index b7f488b..a040d20 100644 --- a/test/integration/jetstream_to_timescale_test.go +++ b/test/integration/jetstream_to_timescale_test.go @@ -10,13 +10,14 @@ import ( "testing" "time" + "caatsm/internal/adapter/dto" "caatsm/internal/adapter/parser" "caatsm/internal/app" - "caatsm/internal/domain" "caatsm/internal/infra/config" loginfra "caatsm/internal/infra/log" natsinfra "caatsm/internal/infra/nats" postgresinfra "caatsm/internal/infra/postgres" + telemetryinfra "caatsm/internal/infra/telemetry" "github.com/jackc/pgx/v5/pgxpool" "github.com/nats-io/nats.go" @@ -80,8 +81,9 @@ func TestJetStreamToTimescaleFlow(t *testing.T) { t.Fatalf("failed to init publisher: %v", err) } - proc := app.NewMessageProcessor(parser.ProvideParser(), repo, publisher, logger) - consumer, err := natsinfra.ProvideConsumer(conn, js, proc, cfg, logger) + telemetryRecorder := telemetryinfra.NewNoop() + proc := app.NewMessageProcessor(parser.ProvideParser(), repo, publisher, telemetryRecorder, logger) + consumer, err := natsinfra.ProvideConsumer(conn, js, proc, cfg, telemetryRecorder, logger) if err != nil { t.Fatalf("failed to init consumer: %v", err) } @@ -131,7 +133,7 @@ NNNN`) SELECT status FROM aviation.telegrams WHERE message_id = $1 LIMIT 1 `, "TMQ2526").Scan(&status) if err == nil { - if status == string(model.MessageStatusParsed) { + if status == string(dto.MessageStatusParsed) { return } t.Logf("message persisted with status=%s, waiting for parsed", status) @@ -275,7 +277,7 @@ func buildTestConfig(natsURL, pgURL string) *config.Config { } func applyDDL(ctx context.Context, pool *pgxpool.Pool) error { - ddlPath := filepath.Join("..", "..", "internal", "repository", "telegrams.ddl") + ddlPath := filepath.Join("..", "..", "internal", "infra", "postgres", "telegrams.ddl") bytes, err := os.ReadFile(ddlPath) if err != nil { return fmt.Errorf("read ddl: %w", err)