diff --git a/Makefile b/Makefile index c5e7eb5..8b6ee0b 100644 --- a/Makefile +++ b/Makefile @@ -38,9 +38,15 @@ run-local: ## Run receiver directly via go run @GO_ENV=$(GO_ENV) go run $(CMD) listen .PHONY: test -test: ## Run unit tests - @echo "Running go test..." - @go test ./... +test: ## Run unit tests (Ginkgo) + @command -v ginkgo >/dev/null || (echo "Please install ginkgo (go install github.com/onsi/ginkgo/v2/ginkgo@latest)"; exit 1) + @echo "Running Ginkgo unit test suites..." + @ginkgo -r ./cmd ./internal + +.PHONY: test-int +test-int: ## Run integration tests (requires Docker) + @echo "Running integration tests..." + @GO_ENV=$(GO_ENV) go test -tags=integration ./test/integration/... .PHONY: test-int test-int: ## Run integration tests (requires Docker) @@ -48,9 +54,12 @@ test-int: ## Run integration tests (requires Docker) @GO_ENV=$(GO_ENV) go test -tags=integration ./test/integration/... .PHONY: test-ginkgo -test-ginkgo: ## Run ginkgo test suites - @command -v ginkgo >/dev/null || (echo "Please install ginkgo (go install github.com/onsi/ginkgo/v2/ginkgo@latest)"; exit 1) - @ginkgo -r -v +test-ginkgo: test ## Alias for test (Ginkgo) + +.PHONY: test-all +test-all: ## Run unit tests (Ginkgo) and integration tests + @$(MAKE) test + @$(MAKE) test-int .PHONY: coverage coverage: ## Run coverage and generate report diff --git a/README.md b/README.md index 01d9cef..54bb08b 100644 --- a/README.md +++ b/README.md @@ -318,19 +318,19 @@ Dependencies are managed using Google Wire. To add a new dependency: 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 is just `go test ./...`. +- **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 (`ginkgo -r ./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`. -| Purpose | Make command | Task command | -|------------------------|---------------------|---------------------| -| Run unit tests | `make test` | `task test` | -| Run integration tests | `make test-int` | `task test-int` | -| Run Ginkgo suites | `make test-ginkgo` | `task test-ginkgo` | -| Generate coverage html | `make coverage` | `task coverage` | -| Lint (golangci-lint) | `make lint` | `task lint` | +| Purpose | Make command | Task command | +|----------------------------|---------------------|---------------------| +| Run unit tests (Ginkgo) | `make test` | `task test` | +| Run integration tests | `make test-int` | `task test-int` | +| Run unit+integration tests | `make test-all` | `task test-all` | +| Generate coverage html | `make coverage` | `task coverage` | +| Lint (golangci-lint) | `make lint` | `task lint` | -> Integration tests need Docker available on the host. Ginkgo or lint targets require the respective binaries (`go install github.com/onsi/ginkgo/v2/ginkgo@latest`, [golangci-lint install guide](https://golangci-lint.run/)). Use `task install-test` to bootstrap Ginkgo tooling. +> Integration tests need Docker available on the host. Ginkgo-based unit tests or lint targets require the respective binaries (`go install github.com/onsi/ginkgo/v2/ginkgo@latest`, [golangci-lint install guide](https://golangci-lint.run/)). Use `task install-test` to bootstrap Ginkgo tooling before running `task test`, `task test-all`, or their `make` equivalents. ## Message Flow diff --git a/Taskfile.yml b/Taskfile.yml index d925918..e8c4e94 100644 --- a/Taskfile.yml +++ b/Taskfile.yml @@ -55,10 +55,15 @@ tasks: GO_ENV=${GO_ENV:-dev} go run {{.cmd}} listen test: - desc: Run go test ./... + desc: Run Ginkgo unit test suites cmds: - - echo "Running go test..." - - go test ./... + - | + if ! command -v ginkgo >/dev/null; then + echo "Install ginkgo (go install github.com/onsi/ginkgo/v2/ginkgo@latest)" + exit 1 + fi + - echo "Running Ginkgo unit test suites..." + - ginkgo -r ./cmd ./internal test-int: desc: Run integration tests (requires Docker) @@ -68,10 +73,15 @@ tasks: GO_ENV=${GO_ENV:-test} go test -tags=integration ./test/integration/... test-ginkgo: - desc: Run ginkgo suites + desc: Alias for Ginkgo unit test suites cmds: - - command -v ginkgo >/dev/null || { echo "Install ginkgo (go install github.com/onsi/ginkgo/v2/ginkgo@latest)"; exit 1; } - - ginkgo -r -v + - task: test + + test-all: + desc: Run unit tests (Ginkgo) and integration tests + cmds: + - task: test + - task: test-int coverage: desc: Generate coverage report (profile + HTML) diff --git a/internal/adapter/mapper/mapper.go b/internal/adapter/mapper/mapper.go index 898f9f6..81505f7 100644 --- a/internal/adapter/mapper/mapper.go +++ b/internal/adapter/mapper/mapper.go @@ -1,13 +1,13 @@ package mapper -import "caatsm/internal/domain" +import "caatsm/internal/model" -// Mapper defines the interface for mapping between domain models and database models +// Mapper defines the interface for mapping between pipeline models and database models type Mapper interface { - // ToDBRow converts a domain.ParsedMessage to a database row representation - ToDBRow(msg *domain.ParsedMessage) ([]interface{}, error) - - // FromDBRow converts a database row to a domain.ParsedMessage - FromDBRow(row []interface{}) (*domain.ParsedMessage, error) + // ToDBRow converts a ParsedTelegram to a database row representation + ToDBRow(msg *model.ParsedTelegram) ([]interface{}, error) + + // FromDBRow converts a database row to a ParsedTelegram + FromDBRow(row []interface{}) (*model.ParsedTelegram, error) } diff --git a/internal/adapter/mapper/telegram.go b/internal/adapter/mapper/telegram.go index c1e4ba4..b093fbc 100644 --- a/internal/adapter/mapper/telegram.go +++ b/internal/adapter/mapper/telegram.go @@ -1,7 +1,7 @@ package mapper import ( - "caatsm/internal/domain" + "caatsm/internal/model" "encoding/json" "fmt" "time" @@ -9,7 +9,7 @@ import ( "github.com/google/uuid" ) -// TelegramMapper maps between domain.ParsedMessage and database rows +// TelegramMapper maps between ParsedTelegram and database rows type TelegramMapper struct{} // NewTelegramMapper creates a new telegram mapper @@ -17,8 +17,8 @@ func NewTelegramMapper() *TelegramMapper { return &TelegramMapper{} } -// ToDBRow converts a domain.ParsedMessage to a database row representation -func (m *TelegramMapper) ToDBRow(msg *domain.ParsedMessage) ([]interface{}, error) { +// ToDBRow converts a ParsedTelegram to a database row representation +func (m *TelegramMapper) ToDBRow(msg *model.ParsedTelegram) ([]interface{}, error) { // Parse UUID var msgUUID uuid.UUID var err error @@ -45,7 +45,7 @@ func (m *TelegramMapper) ToDBRow(msg *domain.ParsedMessage) ([]interface{}, erro status := msg.Status if status == "" { - status = domain.MessageStatusUnknown + status = model.MessageStatusUnknown } return []interface{}{ @@ -69,8 +69,8 @@ func (m *TelegramMapper) ToDBRow(msg *domain.ParsedMessage) ([]interface{}, erro }, nil } -// FromDBRow converts a database row to a domain.ParsedMessage -func (m *TelegramMapper) FromDBRow(row []interface{}) (*domain.ParsedMessage, error) { +// FromDBRow converts a database row to a ParsedTelegram +func (m *TelegramMapper) FromDBRow(row []interface{}) (*model.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{}) (*domain.ParsedMessage, er } } - status := domain.MessageStatusUnknown + status := model.MessageStatusUnknown if rawStatus := toString(row[11]); rawStatus != "" { - status = domain.MessageStatus(rawStatus) + status = model.MessageStatus(rawStatus) } - return &domain.ParsedMessage{ + return &model.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 3b299f5..b7afe43 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/domain" + "caatsm/internal/model" "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 := &domain.ParsedMessage{} + msg := &model.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 := &domain.ParsedMessage{ + original := &model.ParsedTelegram{ Uuid: uuid.NewString(), MessageID: "TMQ1324", DateTime: "150631", @@ -49,7 +49,7 @@ var _ = Describe("TelegramMapper", func() { ParsedAt: now, DispatchedAt: now, NeedDispatch: true, - Status: domain.MessageStatusParsed, + Status: model.MessageStatusParsed, } row, err := mapper.ToDBRow(original) diff --git a/internal/adapter/parser/aviation_parser.go b/internal/adapter/parser/aviation_parser.go index 220612f..000c1f1 100644 --- a/internal/adapter/parser/aviation_parser.go +++ b/internal/adapter/parser/aviation_parser.go @@ -1,7 +1,7 @@ package parser import ( - "caatsm/internal/domain" + "caatsm/internal/model" "caatsm/internal/parsers" ) @@ -13,8 +13,8 @@ func NewAviationParser() *AviationParser { return &AviationParser{} } -// Parse parses a raw message string and returns a ParsedMessage -func (p *AviationParser) Parse(rawText string) (*domain.ParsedMessage, error) { +// 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/adapter/parser/parser.go b/internal/adapter/parser/parser.go index dfaec04..08ecfb2 100644 --- a/internal/adapter/parser/parser.go +++ b/internal/adapter/parser/parser.go @@ -1,9 +1,9 @@ package parser -import "caatsm/internal/domain" +import "caatsm/internal/model" // Parser defines the interface for parsing raw telegram messages type Parser interface { - // Parse parses a raw message string and returns a ParsedMessage - Parse(rawText string) (*domain.ParsedMessage, error) + // Parse parses a raw message string and returns a ParsedTelegram + Parse(rawText string) (*model.ParsedTelegram, error) } diff --git a/internal/adapter/repository.go b/internal/adapter/repository.go index d73d708..f04ded1 100644 --- a/internal/adapter/repository.go +++ b/internal/adapter/repository.go @@ -2,18 +2,18 @@ package adapter import ( "context" - "caatsm/internal/domain" + "caatsm/internal/model" ) // Repository defines the interface for message persistence type Repository interface { // InsertOne inserts a single telegram message - InsertOne(ctx context.Context, msg *domain.ParsedMessage) error + InsertOne(ctx context.Context, msg *model.ParsedTelegram) error // InsertBatch inserts multiple telegram messages in a batch - InsertBatch(ctx context.Context, msgs []*domain.ParsedMessage) error + InsertBatch(ctx context.Context, msgs []*model.ParsedTelegram) error // InsertRaw captures an unparsed or failed telegram for later analysis. - InsertRaw(ctx context.Context, msg *domain.ParsedMessage) error + InsertRaw(ctx context.Context, msg *model.ParsedTelegram) error } diff --git a/internal/app/processor.go b/internal/app/processor.go index ac9b04c..755f2ae 100644 --- a/internal/app/processor.go +++ b/internal/app/processor.go @@ -3,7 +3,7 @@ package app import ( "caatsm/internal/adapter" "caatsm/internal/adapter/parser" - "caatsm/internal/domain" + "caatsm/internal/model" obslogging "caatsm/internal/observability/logging" obsmetrics "caatsm/internal/observability/metrics" "context" @@ -27,6 +27,17 @@ type MessageProcessor struct { logger *zap.Logger } +// ProcessingStatus represents the outcome of the processing pipeline +// (persistence, publishing, etc.), independent from the parsing status +// captured in model.MessageStatus. +type ProcessingStatus string + +const ( + ProcessingStatusOK ProcessingStatus = "ok" + ProcessingStatusPersistFailed ProcessingStatus = "persist_failed" + ProcessingStatusPublishFailed ProcessingStatus = "publish_failed" +) + var ( appMeter = otel.Meter("caatsm/app") messageStatusAttrKey = attribute.Key("message.status") @@ -83,10 +94,10 @@ func (p *MessageProcessor) Handle(ctx context.Context, raw []byte, msgID string) parsed, parseErr := p.parser.Parse(string(raw)) if parsed == nil { - parsed = domain.NewParsedMessage() + parsed = model.NewParsedTelegram() parsed.Content = string(raw) parsed.ErrorReason = "parser returned nil" - parsed.Status = domain.MessageStatusBodyError + parsed.Status = model.MessageStatusBodyError parseErr = fmt.Errorf("parser returned nil") } @@ -103,11 +114,11 @@ func (p *MessageProcessor) Handle(ctx context.Context, raw []byte, msgID string) if parsed.ParsedAt.IsZero() { parsed.ParsedAt = time.Now() } - if parsed.Status == domain.MessageStatusUnknown { + if parsed.Status == model.MessageStatusUnknown { if parseErr == nil { - parsed.Status = domain.MessageStatusParsed + parsed.Status = model.MessageStatusParsed } else { - parsed.Status = domain.MessageStatusBodyError + parsed.Status = model.MessageStatusBodyError } } @@ -166,7 +177,6 @@ func (p *MessageProcessor) Handle(ctx context.Context, raw []byte, msgID string) if err := p.repository.InsertOne(ctx, parsed); err != nil { span.RecordError(err) span.SetStatus(codes.Error, err.Error()) - parsed.Status = domain.MessageStatusRepositoryFail latency := parsed.ParsedAt.Sub(receivedAt) parseLatencyHistogram.Record(ctx, float64(latency.Milliseconds()), metric.WithAttributes( @@ -189,7 +199,6 @@ func (p *MessageProcessor) Handle(ctx context.Context, raw []byte, msgID string) ) pubSpan.RecordError(err) pubSpan.SetStatus(codes.Error, err.Error()) - parsed.Status = domain.MessageStatusPublishFail parsed.ErrorReason = err.Error() messagePublishFailCounter.Add(ctx, 1, metric.WithAttributes( @@ -224,7 +233,7 @@ func (p *MessageProcessor) Handle(ctx context.Context, raw []byte, msgID string) return nil } -func (p *MessageProcessor) persistRaw(ctx context.Context, msg *domain.ParsedMessage) { +func (p *MessageProcessor) persistRaw(ctx context.Context, msg *model.ParsedTelegram) { if msg == nil || p.repository == nil { return } @@ -261,7 +270,7 @@ func truncateContent(content string, limit int) string { return content[:limit-3] + "..." } -func recordProcessedMetric(ctx context.Context, msg *domain.ParsedMessage, elapsed time.Duration) { +func recordProcessedMetric(ctx context.Context, msg *model.ParsedTelegram, elapsed time.Duration) { if msg == nil { return } diff --git a/internal/app/processor_test.go b/internal/app/processor_test.go index f4f59a4..e4b6a91 100644 --- a/internal/app/processor_test.go +++ b/internal/app/processor_test.go @@ -8,7 +8,7 @@ import ( "caatsm/internal/adapter" "caatsm/internal/adapter/parser" - "caatsm/internal/domain" + "caatsm/internal/model" "github.com/google/uuid" . "github.com/onsi/ginkgo/v2" @@ -51,7 +51,7 @@ var _ = Describe("MessageProcessor", func() { It("preserves UUIDs and appends nats message id comment", func() { originalUUID := uuid.NewString() - parserStub.value = &domain.ParsedMessage{Uuid: originalUUID, Parsed: true, Status: domain.MessageStatusParsed} + parserStub.value = &model.ParsedTelegram{Uuid: originalUUID, Parsed: true, Status: model.MessageStatusParsed} Expect(proc.Handle(ctx, []byte("payload"), "msg-123")).To(Succeed()) @@ -62,7 +62,7 @@ var _ = Describe("MessageProcessor", func() { }) It("treats publisher failures as permanent and stores raw entries", func() { - parserStub.value = &domain.ParsedMessage{Parsed: true, Status: domain.MessageStatusParsed} + parserStub.value = &model.ParsedTelegram{Parsed: true, Status: model.MessageStatusParsed} pub.err = errors.New("publish failed") err := proc.Handle(ctx, []byte("payload"), "id-3") @@ -70,14 +70,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(domain.MessageStatusPublishFail)) + Expect(repo.lastRaw().Status).To(Equal(model.MessageStatusParsed)) + Expect(repo.lastRaw().ErrorReason).To(ContainSubstring("publish failed")) }) It("sets timestamps when missing", func() { - parserStub.value = &domain.ParsedMessage{ + parserStub.value = &model.ParsedTelegram{ Uuid: uuid.NewString(), Parsed: true, - Status: domain.MessageStatusParsed, + Status: model.MessageStatusParsed, } pub.err = nil @@ -95,10 +96,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 = &domain.ParsedMessage{ + parserStub.value = &model.ParsedTelegram{ Uuid: uuid.NewString(), Parsed: true, - Status: domain.MessageStatusParsed, + Status: model.MessageStatusParsed, ReceivedAt: received, ParsedAt: parsedAt, } @@ -112,10 +113,10 @@ var _ = Describe("MessageProcessor", func() { core, logs := observer.New(zap.WarnLevel) logger := zap.New(core) parserStub = &stubParser{ - value: &domain.ParsedMessage{ + value: &model.ParsedTelegram{ Content: strings.Repeat("x", 1024), Parsed: false, - Status: domain.MessageStatusBodyError, + Status: model.MessageStatusBodyError, ErrorReason: "parse failure", }, err: errors.New("parse failure"), @@ -148,22 +149,22 @@ func newTestProcessor(p parser.Parser, repo adapter.Repository, pub adapter.Publ } type stubParser struct { - value *domain.ParsedMessage + value *model.ParsedTelegram err error } -func (s *stubParser) Parse(rawText string) (*domain.ParsedMessage, error) { +func (s *stubParser) Parse(rawText string) (*model.ParsedTelegram, error) { return s.value, s.err } type stubRepository struct { - inserted []*domain.ParsedMessage - raw []*domain.ParsedMessage + inserted []*model.ParsedTelegram + raw []*model.ParsedTelegram err error rawErr error } -func (s *stubRepository) InsertOne(ctx context.Context, msg *domain.ParsedMessage) error { +func (s *stubRepository) InsertOne(ctx context.Context, msg *model.ParsedTelegram) error { if s.err != nil { return s.err } @@ -171,11 +172,11 @@ func (s *stubRepository) InsertOne(ctx context.Context, msg *domain.ParsedMessag return nil } -func (s *stubRepository) InsertBatch(ctx context.Context, msgs []*domain.ParsedMessage) error { +func (s *stubRepository) InsertBatch(ctx context.Context, msgs []*model.ParsedTelegram) error { return errors.New("not implemented") } -func (s *stubRepository) InsertRaw(ctx context.Context, msg *domain.ParsedMessage) error { +func (s *stubRepository) InsertRaw(ctx context.Context, msg *model.ParsedTelegram) error { if s.rawErr != nil { return s.rawErr } @@ -183,14 +184,14 @@ func (s *stubRepository) InsertRaw(ctx context.Context, msg *domain.ParsedMessag return nil } -func (s *stubRepository) last() *domain.ParsedMessage { +func (s *stubRepository) last() *model.ParsedTelegram { if len(s.inserted) == 0 { return nil } return s.inserted[len(s.inserted)-1] } -func (s *stubRepository) lastRaw() *domain.ParsedMessage { +func (s *stubRepository) lastRaw() *model.ParsedTelegram { if len(s.raw) == 0 { return nil } diff --git a/internal/domain/aviation.go b/internal/domain/aviation.go index ad5aef4..4c556b0 100644 --- a/internal/domain/aviation.go +++ b/internal/domain/aviation.go @@ -1,7 +1,5 @@ package domain -import "time" - // An aviation message typically contains various fields that are crucial for air traffic management and communication. //These fields include identifiers, date and time, priority indicators, addresses, //and additional information such as call signs, flight plans, route details, altitude, speed, position, @@ -84,52 +82,6 @@ DispatchedAt: time.Time{}. NeedDispatch: false. */ -// ParsedMessage holds the parsed data from an aviation message -type MessageStatus string - -const ( - MessageStatusUnknown MessageStatus = "unknown" - MessageStatusParsed MessageStatus = "parsed" - MessageStatusHeaderError MessageStatus = "header_error" - MessageStatusBodyError MessageStatus = "body_error" - MessageStatusRepositoryFail MessageStatus = "repository_error" - MessageStatusPublishFail MessageStatus = "publish_error" -) - -// ParsedMessage holds the parsed data from an aviation message -type ParsedMessage struct { - // StartIndicator string `json:"startIndicator"` // 电报开始标识: The start of the message indicator (e.g., 'ZCZC'). - Uuid string `json:"uuid"` - MessageID string `json:"messageId"` // 信息ID: The message ID (e.g., 'TMQ1324'). - DateTime string `json:"dateTime"` // 日期时间: The date and time of the message (e.g., '150631'). - PriorityIndicator string `json:"priorityIndicator"` // 优先级标识: The priority level of the message (e.g., 'FF'). - PrimaryAddress string `json:"primaryAddress"` // 主要地址: The primary recipient address (e.g., 'ZBTJZPZX'). - SecondaryAddresses string `json:"secondaryAddresses,omitempty"` // 次要地址: Additional recipient addresses (space-separated string such as "150630 ZBACZQZX"). - Originator string `json:"originator,omitempty"` // 发件人: The sender of the message. - OriginatorDateTime string `json:"originatorDateTime,omitempty"` // 发件日期时间: The date and time when the originator sent the message. - Category string `json:"category,omitempty"` // 类别: The category of the message. - Body string // 正文和页脚: The body and footer of the message (e.g., 'CALLSIGN/ABC123\nFPL/AB1234-AB\n...'). - Content string `json:"content,omitempty"` // 正文: The body of the message. - BodyData interface{} `json:"bodyData,omitempty"` // 正文数据: Parsed body data. - ReceivedAt time.Time `json:"receivedAt"` // 接收时间: The time when the message was received. - ParsedAt time.Time `json:"parsedAt,omitempty"` // 解析时间: The time when the message was parsed. - DispatchedAt time.Time `json:"dispatchedAt,omitempty"` // 分发时间: The time when the message was dispatched. - NeedDispatch bool `json:"needDispatch"` // 需要分发: Indicates if the message needs to be dispatched. - Parsed bool `json:"parsed"` // 解析: Indicates if the message has been parsed. - Comments string `json:"comments,omitempty"` // 备注: Additional comments. - Status MessageStatus - ErrorReason string `json:"errorReason,omitempty"` -} - -// NewParsedMessage initializes a ParsedMessage with default values -func NewParsedMessage() *ParsedMessage { - return &ParsedMessage{ - // SecondaryAddresses: []string{}, - Parsed: false, - Status: MessageStatusUnknown, - } -} - -func (message *ParsedMessage) ToString() string { - return message.MessageID + " " + message.Category + " " + message.Originator -} +// NOTE: Parsed telegram pipeline structures (ParsedTelegram, MessageStatus, etc.) +// have been moved to the internal/model 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 index 97264b2..54a9dcc 100644 --- a/internal/iface/interface.go +++ b/internal/iface/interface.go @@ -1,9 +1,6 @@ package iface -import ( - "caatsm/internal/domain" - "caatsm/internal/infra/config" -) +import "caatsm/internal/infra/config" type MessageHandler interface { HandleMessage(msg []byte, id string) error @@ -18,5 +15,5 @@ type MessageSubscriber interface { } type MessageRepository interface { - CreateNew(message *domain.ParsedMessage) error + CreateNew(message interface{}) error } diff --git a/internal/infra/nats/publisher.go b/internal/infra/nats/publisher.go index 21e5fcd..89b4ee7 100644 --- a/internal/infra/nats/publisher.go +++ b/internal/infra/nats/publisher.go @@ -2,8 +2,8 @@ package nats import ( "caatsm/internal/adapter" - "caatsm/internal/domain" "caatsm/internal/infra/config" + "caatsm/internal/model" "encoding/json" "fmt" "github.com/google/uuid" @@ -49,7 +49,7 @@ func (p *Publisher) Publish(message interface{}) error { jsMsg.Data = messageBytes switch typed := message.(type) { - case *domain.ParsedMessage: + case *model.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 ae77522..ce65f7b 100644 --- a/internal/infra/postgres/repository.go +++ b/internal/infra/postgres/repository.go @@ -3,7 +3,7 @@ package postgres import ( "caatsm/internal/adapter" "caatsm/internal/adapter/mapper" - "caatsm/internal/domain" + "caatsm/internal/model" obsmetrics "caatsm/internal/observability/metrics" "context" "encoding/json" @@ -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 *domain.ParsedMessage) error { +func (r *Repository) InsertOne(ctx context.Context, msg *model.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 *domain.ParsedMessage) e } // InsertBatch inserts multiple telegram messages in a batch using CopyFrom -func (r *Repository) InsertBatch(ctx context.Context, msgs []*domain.ParsedMessage) error { +func (r *Repository) InsertBatch(ctx context.Context, msgs []*model.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 []*domain.ParsedMessa } // InsertRaw inserts a failed telegram into aviation.telegrams_raw for post-processing. -func (r *Repository) InsertRaw(ctx context.Context, msg *domain.ParsedMessage) error { +func (r *Repository) InsertRaw(ctx context.Context, msg *model.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/model/telegram.go b/internal/model/telegram.go new file mode 100644 index 0000000..510aee3 --- /dev/null +++ b/internal/model/telegram.go @@ -0,0 +1,52 @@ +package model + +import ( + "time" +) + +// MessageStatus represents the parsing status of a telegram. +// It is intentionally decoupled from infrastructure concerns (e.g. DB or publish failures) +// so that domain parsing state can be reasoned about independently. +type MessageStatus string + +const ( + MessageStatusUnknown MessageStatus = "unknown" + MessageStatusParsed MessageStatus = "parsed" + MessageStatusHeaderError MessageStatus = "header_error" + MessageStatusBodyError MessageStatus = "body_error" +) + +// ParsedTelegram holds the parsed data from an aviation message. +// It is a transport-oriented model used by the application pipeline (parser, +// persistence, publishing), and may embed domain-specific body structures +// (e.g. *domain.FPL, *domain.DEP) in BodyData. +type ParsedTelegram struct { + Uuid string `json:"uuid"` + MessageID string `json:"messageId"` + DateTime string `json:"dateTime"` + PriorityIndicator string `json:"priorityIndicator"` + PrimaryAddress string `json:"primaryAddress"` + SecondaryAddresses string `json:"secondaryAddresses,omitempty"` + Originator string `json:"originator,omitempty"` + OriginatorDateTime string `json:"originatorDateTime,omitempty"` + Category string `json:"category,omitempty"` + Body string `json:"body"` + Content string `json:"content,omitempty"` + BodyData interface{} `json:"bodyData,omitempty"` + ReceivedAt time.Time `json:"receivedAt"` + ParsedAt time.Time `json:"parsedAt,omitempty"` + DispatchedAt time.Time `json:"dispatchedAt,omitempty"` + NeedDispatch bool `json:"needDispatch"` + Parsed bool `json:"parsed"` + Comments string `json:"comments,omitempty"` + Status MessageStatus + ErrorReason string `json:"errorReason,omitempty"` +} + +// NewParsedTelegram initializes a ParsedTelegram with default values. +func NewParsedTelegram() *ParsedTelegram { + return &ParsedTelegram{ + Parsed: false, + Status: MessageStatusUnknown, + } +} diff --git a/internal/parsers/aviation_parser.go b/internal/parsers/aviation_parser.go index facae16..aa8cc93 100644 --- a/internal/parsers/aviation_parser.go +++ b/internal/parsers/aviation_parser.go @@ -2,6 +2,7 @@ package parsers import ( "caatsm/internal/domain" + "caatsm/internal/model" "caatsm/pkg/utils" "errors" "fmt" @@ -199,33 +200,65 @@ func (parser *BodyParser) createBodyData(data map[string]string) (string, interf } } -func Parse(rawText string) (*domain.ParsedMessage, error) { - message, err := ParseHeader(rawText) +func Parse(rawText string) (*model.ParsedTelegram, error) { + header, err := ParseHeader(rawText) if err != nil { - msg := domain.NewParsedMessage() + msg := model.NewParsedTelegram() msg.Content = rawText msg.Comments = err.Error() msg.ErrorReason = err.Error() - msg.Status = domain.MessageStatusHeaderError + msg.Status = model.MessageStatusHeaderError return msg, fmt.Errorf("%w: %w", ErrHeaderParse, err) } - bodyParser := NewBodyParser(message.Body) + bodyParser := NewBodyParser(header.Body) category, bodyData, bodyErr := bodyParser.Parse() - message.Category = category - message.ParsedAt = time.Now() + header.Category = category + header.ParsedAt = time.Now() if bodyErr != nil { - message.Comments = bodyErr.Error() - message.ErrorReason = bodyErr.Error() - message.Status = domain.MessageStatusBodyError - return &message, fmt.Errorf("%w: %w", ErrBodyParse, bodyErr) + return &model.ParsedTelegram{ + MessageID: header.MessageID, + DateTime: header.DateTime, + PriorityIndicator: header.PriorityIndicator, + PrimaryAddress: header.PrimaryAddress, + SecondaryAddresses: header.SecondaryAddresses, + Originator: header.Originator, + OriginatorDateTime: header.OriginatorDateTime, + Category: header.Category, + Body: header.Body, + Content: header.Content, + ReceivedAt: header.ReceivedAt, + ParsedAt: header.ParsedAt, + Parsed: false, + Comments: bodyErr.Error(), + Status: model.MessageStatusBodyError, + ErrorReason: bodyErr.Error(), + }, fmt.Errorf("%w: %w", ErrBodyParse, bodyErr) } - message.Parsed = true - message.Status = domain.MessageStatusParsed - message.BodyData = bodyData - message.Uuid = uuid.New().String() - return &message, nil + + parsed := &model.ParsedTelegram{ + MessageID: header.MessageID, + DateTime: header.DateTime, + PriorityIndicator: header.PriorityIndicator, + PrimaryAddress: header.PrimaryAddress, + SecondaryAddresses: header.SecondaryAddresses, + Originator: header.Originator, + OriginatorDateTime: header.OriginatorDateTime, + Category: header.Category, + Body: header.Body, + Content: header.Content, + BodyData: bodyData, + ReceivedAt: header.ReceivedAt, + ParsedAt: header.ParsedAt, + Parsed: true, + Status: model.MessageStatusParsed, + ErrorReason: "", + } + + parsed.Uuid = uuid.New().String() + + return parsed, nil } func cleanMessage(text string) string { @@ -241,25 +274,42 @@ func cleanMessage(text string) string { return "" } -func ParseHeader(fullMessage string) (domain.ParsedMessage, error) { +// ParseHeader parses only the header portion of the message and returns a lightweight struct +// with header fields and body content. It is used internally by the aviation parser. +type Header struct { + MessageID string + DateTime string + PriorityIndicator string + PrimaryAddress string + SecondaryAddresses string + Originator string + OriginatorDateTime string + Category string + Content string + Body string + ReceivedAt time.Time + ParsedAt time.Time +} + +func ParseHeader(fullMessage string) (Header, error) { log := utils.GetSugaredLogger() cleaned := cleanMessage(fullMessage) lines := strings.Split(cleaned, "\n") if len(lines) < 3 { log.Warnf("invalid message format: %s", fullMessage) - return domain.ParsedMessage{Content: fullMessage}, fmt.Errorf("invalid message format: %s", fullMessage) + return Header{Content: fullMessage}, fmt.Errorf("invalid message format: %s", fullMessage) } _, messageID, dateTime, err := parseStartIndicator(lines[0]) if err != nil { - return domain.ParsedMessage{Content: fullMessage}, err + return Header{Content: fullMessage}, err } priorityIndicator, primaryAddress := parsePriorityAndPrimary(lines[1]) secondaryAddresses, originator, originatorDateTime, body := parseRemainingLines(lines[2:]) - return domain.ParsedMessage{ + return Header{ MessageID: messageID, DateTime: dateTime, PriorityIndicator: priorityIndicator, diff --git a/test/integration/jetstream_to_timescale_test.go b/test/integration/jetstream_to_timescale_test.go index d2fc334..b7f488b 100644 --- a/test/integration/jetstream_to_timescale_test.go +++ b/test/integration/jetstream_to_timescale_test.go @@ -131,7 +131,7 @@ NNNN`) SELECT status FROM aviation.telegrams WHERE message_id = $1 LIMIT 1 `, "TMQ2526").Scan(&status) if err == nil { - if status == string(domain.MessageStatusParsed) { + if status == string(model.MessageStatusParsed) { return } t.Logf("message persisted with status=%s, waiting for parsed", status)