diff --git a/README.md b/README.md index 7cde5bb..17bd3ca 100644 --- a/README.md +++ b/README.md @@ -80,6 +80,8 @@ Configuration is loaded from TOML files and environment variables. The configura url = "nats://localhost:4222" stream = "TELEGRAM" consumer = "telegram-consumer" +client = "serial-client" +cluster = "tele-cluster" [nats.stream_limits] max_msgs = 100000 @@ -93,10 +95,16 @@ replicas = 1 max_deliver = 5 ack_wait = "30s" max_ack_pending = 1024 +deliver_policy = "all" # all,new,last,last_per_subject,sequence,time +replay_policy = "instant" # instant or original +backoff = ["5s", "30s", "2m"] # optional JetStream redelivery delays +start_sequence = 0 +start_time = "" [subscription] # Optional. Defaults to "telegram.>" when omitted. topic = "telegram.serial" +queue_group = "tele-queue" [publisher] topic = "telegram.json" @@ -115,6 +123,11 @@ monitor_interval = "30s" level = "info" format = "json" +[telemetry] +enabled = false +endpoint = "http://otel-collector:4318" +insecure = true + ### Timeouts and Ack Wait `[timeouts]` is optional, but if you plan to tune JetStream redelivery you should set `timeouts.ack_wait` and/or `[nats.consumer].ack_wait`. When neither is specified the application defaults both values to `30s`, ensuring predictable redelivery timing. @@ -168,10 +181,48 @@ task run-dev ./bin/receiver listen --help Flags: - -n, --nats string Nats server address (default: "nats://localhost:4222") - -t, --topic string Nats topic to listen to (default: "telegram.serial") + -n, --nats-url string NATS server address + -t, --subject string NATS subject to listen to + --stream string JetStream stream name + --consumer string JetStream durable consumer + --publisher-topic string Subject used by the publisher + --postgres-url string PostgreSQL connection URL + --log-level string Logger level (debug|info|warn|error) + --replay-from string Deliver policy override (all|new|last|seq:|time:) + --ack-wait duration Ack wait override (e.g. 45s) + --telemetry-enabled Enable OpenTelemetry exporters + --telemetry-endpoint string + OTLP collector endpoint + --telemetry-insecure Send OTLP traffic without TLS ``` +Critical overrides stay available through CLI flags; advanced tuning such as stream retention, consumer backoff, and copy counts are configured via the TOML file or `CAATSM_` environment variables. + +| CLI flag | Config key | Purpose | +|---------------------|------------------------|----------------------------------------| +| `--nats-url` | `nats.url` | Point to a different NATS cluster | +| `--subject` | `subscription.topic` | Change the subscribed subject filter | +| `--stream` | `nats.stream` | Bind to another JetStream stream | +| `--consumer` | `nats.consumer` | Override the durable consumer name | +| `--publisher-topic` | `publisher.topic` | Publish parsed output to a new subject | +| `--postgres-url` | `postgres.url` | Redirect persistence to another DB | +| `--log-level` | `log.level` | Adjust runtime logging verbosity | +| `--telemetry-*` | `telemetry.*` | Toggle tracing/metrics exporters | + +#### Replay & Backoff + +- `--replay-from seq:12345` replays from a specific JetStream sequence, while `--replay-from time:2024-11-15T08:00:00Z` starts at a timestamp. +- Configure server-side retry delays with `[nats.consumer].backoff = ["5s", "30s", "2m"]`; each duration becomes the delay before the next delivery attempt. +- Combine `backoff` with `--ack-wait` to increase acknowledgement windows (e.g., `--ack-wait 2m`). + +### Telemetry + +- Enable tracing/metrics via `[telemetry] enabled = true` and set `endpoint` to your OTLP/HTTP collector (e.g., `http://otel-collector:4318`). +- CLI overrides: + - `--telemetry-enabled` flips the feature on/off. + - `--telemetry-endpoint` and `--telemetry-insecure` adjust the OTLP HTTP endpoint and TLS behavior. +- When enabled the app emits OpenTelemetry traces (parser/repository/publisher spans) and JetStream metrics (ack pending, deliveries) for dashboards and alerts. + ## Development ### Project Structure @@ -196,6 +247,8 @@ Dependencies are managed using Google Wire. To add a new dependency: 2. Add it to `pkg/di/wire.go` 3. Run `wire ./pkg/di` to regenerate `wire_gen.go` +> If the `wire` binary is missing, install it with `go install github.com/google/wire/cmd/wire@v0.7.0` and ensure `$GOPATH/bin` is on your `PATH` (or run it directly via the absolute path). + ### Testing ```bash @@ -204,6 +257,9 @@ go test ./... # Run tests with coverage task coverage + +# Run all Ginkgo suites (requires go install github.com/onsi/ginkgo/v2/ginkgo@latest) +ginkgo -r ``` ## Message Flow @@ -216,6 +272,19 @@ task coverage 3. **ACK/NAK** is sent based on processing success/failure 4. **Retry Logic** handles transient failures automatically +### Failure Buckets + +Messages that cannot be parsed or fail to publish are written to `aviation.telegrams_raw` with a status: + +| Status | Description | +|-----------------------|--------------------------------------------------| +| `parsed` | Successfully parsed and stored | +| `header_error` | Header invalid (missing start indicator, etc.) | +| `body_error` | Body pattern did not match any known format | +| `publish_error` | Downstream publisher returned an error | + +Each entry stores the raw payload, received timestamp, and metadata to aid replay or manual inspection. + ## Message Parsing The system supports parsing of aviation telegram messages in the standard ICAO format. All messages follow a common header structure, followed by a message body that varies by message type. diff --git a/cmd/main/main.go b/cmd/main/main.go index fe8932a..ac64da5 100644 --- a/cmd/main/main.go +++ b/cmd/main/main.go @@ -8,10 +8,21 @@ import ( "fmt" "os" "os/signal" + "strconv" + "strings" "syscall" "time" "github.com/urfave/cli/v2" + "go.opentelemetry.io/otel" + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetrichttp" + "go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp" + "go.opentelemetry.io/otel/propagation" + sdkmetric "go.opentelemetry.io/otel/sdk/metric" + "go.opentelemetry.io/otel/sdk/resource" + sdktrace "go.opentelemetry.io/otel/sdk/trace" + semconv "go.opentelemetry.io/otel/semconv/v1.26.0" ) func main() { @@ -32,19 +43,60 @@ func setupApp() *cli.App { Usage: "Listen to nats messages", Flags: []cli.Flag{ &cli.StringFlag{ - Name: "nats", + Name: "nats-url", Aliases: []string{"n"}, - Usage: "Nats server address", - Value: "nats://localhost:4222", + Usage: "NATS server address", EnvVars: []string{"NATS_SERVER"}, }, &cli.StringFlag{ - Name: "topic", + Name: "subject", Aliases: []string{"t"}, - Usage: "Nats topic to listen to", - Value: "telegram.serial", + Usage: "NATS subject to listen to", EnvVars: []string{"NATS_SUBJECT"}, }, + &cli.StringFlag{ + Name: "stream", + Usage: "NATS JetStream stream name", + }, + &cli.StringFlag{ + Name: "consumer", + Usage: "NATS JetStream durable consumer", + }, + &cli.StringFlag{ + Name: "publisher-topic", + Usage: "Subject used by the publisher", + }, + &cli.StringFlag{ + Name: "postgres-url", + Usage: "PostgreSQL connection URL", + EnvVars: []string{ + "POSTGRES_URL", + }, + }, + &cli.StringFlag{ + Name: "log-level", + Usage: "Logger level (debug, info, warn, error)", + }, + &cli.StringFlag{ + Name: "replay-from", + Usage: "Override deliver policy: all|new|last|seq:|time:", + }, + &cli.DurationFlag{ + Name: "ack-wait", + Usage: "Override consumer ack wait duration", + }, + &cli.BoolFlag{ + Name: "telemetry-enabled", + Usage: "Enable OpenTelemetry exporters", + }, + &cli.StringFlag{ + Name: "telemetry-endpoint", + Usage: "OpenTelemetry collector OTLP endpoint", + }, + &cli.BoolFlag{ + Name: "telemetry-insecure", + Usage: "Send OTLP data without TLS", + }, }, Action: executeListen, }, @@ -58,12 +110,16 @@ func executeListen(c *cli.Context) error { return fmt.Errorf("failed to load config: %w", err) } - if flagURL := c.String("nats"); flagURL != "" { - cfg.NATS.URL = flagURL - } + applyCLIOverrides(cfg, c) - if flagTopic := c.String("topic"); flagTopic != "" { - cfg.Subscription.Topic = flagTopic + shutdownTelemetry := func(context.Context) error { return nil } + if cfg.Telemetry.Enabled { + var telErr error + shutdownTelemetry, telErr = initTelemetry(context.Background(), cfg) + if telErr != nil { + return fmt.Errorf("failed to initialize telemetry: %w", telErr) + } + defer shutdownTelemetry(context.Background()) } // Initialize dependencies using Wire @@ -123,3 +179,162 @@ func executeListen(c *cli.Context) error { return runErr } + +func applyCLIOverrides(cfg *config.Config, c *cli.Context) { + if cfg == nil || c == nil { + return + } + if flagURL := c.String("nats-url"); flagURL != "" { + cfg.NATS.URL = flagURL + } + if flagTopic := c.String("subject"); flagTopic != "" { + cfg.Subscription.Topic = flagTopic + } + if stream := c.String("stream"); stream != "" { + cfg.NATS.Stream = stream + } + if consumer := c.String("consumer"); consumer != "" { + cfg.NATS.Consumer = consumer + } + if publisherTopic := c.String("publisher-topic"); publisherTopic != "" { + cfg.Publisher.Topic = publisherTopic + } + if pgURL := c.String("postgres-url"); pgURL != "" { + cfg.Postgres.URL = pgURL + } + if logLevel := c.String("log-level"); logLevel != "" { + cfg.Log.Level = logLevel + } + if replay := c.String("replay-from"); replay != "" { + applyReplayOverride(cfg, replay) + } + if c.IsSet("ack-wait") { + if ack := c.Duration("ack-wait"); ack > 0 { + cfg.Timeouts.AckWait = ack + cfg.NATS.ConsumerRules.AckWait = ack + } + } + if c.IsSet("telemetry-enabled") { + cfg.Telemetry.Enabled = c.Bool("telemetry-enabled") + } + if endpoint := c.String("telemetry-endpoint"); endpoint != "" { + cfg.Telemetry.Endpoint = endpoint + } + if c.IsSet("telemetry-insecure") { + cfg.Telemetry.Insecure = c.Bool("telemetry-insecure") + } +} + +func applyReplayOverride(cfg *config.Config, value string) { + if cfg == nil { + return + } + lower := strings.ToLower(strings.TrimSpace(value)) + switch { + case lower == "new": + cfg.NATS.ConsumerRules.DeliverPolicy = "new" + cfg.NATS.ConsumerRules.StartSequence = 0 + cfg.NATS.ConsumerRules.StartTime = "" + case lower == "all": + cfg.NATS.ConsumerRules.DeliverPolicy = "all" + cfg.NATS.ConsumerRules.StartSequence = 0 + cfg.NATS.ConsumerRules.StartTime = "" + case lower == "last": + cfg.NATS.ConsumerRules.DeliverPolicy = "last" + cfg.NATS.ConsumerRules.StartSequence = 0 + cfg.NATS.ConsumerRules.StartTime = "" + case lower == "last_per_subject": + cfg.NATS.ConsumerRules.DeliverPolicy = "last_per_subject" + cfg.NATS.ConsumerRules.StartSequence = 0 + cfg.NATS.ConsumerRules.StartTime = "" + case strings.HasPrefix(lower, "seq:"): + seqStr := strings.TrimPrefix(lower, "seq:") + if seq, err := strconv.ParseUint(seqStr, 10, 64); err == nil { + cfg.NATS.ConsumerRules.DeliverPolicy = "sequence" + cfg.NATS.ConsumerRules.StartSequence = seq + cfg.NATS.ConsumerRules.StartTime = "" + } + case strings.HasPrefix(lower, "time:"): + ts := strings.TrimSpace(value[5:]) + if _, err := time.Parse(time.RFC3339, ts); err == nil { + cfg.NATS.ConsumerRules.DeliverPolicy = "time" + cfg.NATS.ConsumerRules.StartSequence = 0 + cfg.NATS.ConsumerRules.StartTime = ts + } + } +} + +func initTelemetry(ctx context.Context, cfg *config.Config) (func(context.Context) error, error) { + if cfg == nil || !cfg.Telemetry.Enabled { + return func(context.Context) error { return nil }, nil + } + if cfg.Telemetry.Endpoint == "" { + return nil, fmt.Errorf("telemetry endpoint is required when telemetry.enabled=true") + } + + traceOpts := []otlptracehttp.Option{ + otlptracehttp.WithEndpoint(cfg.Telemetry.Endpoint), + otlptracehttp.WithURLPath("/v1/traces"), + } + metricOpts := []otlpmetrichttp.Option{ + otlpmetrichttp.WithEndpoint(cfg.Telemetry.Endpoint), + otlpmetrichttp.WithURLPath("/v1/metrics"), + } + if cfg.Telemetry.Insecure { + traceOpts = append(traceOpts, otlptracehttp.WithInsecure()) + metricOpts = append(metricOpts, otlpmetrichttp.WithInsecure()) + } + + traceExporter, err := otlptracehttp.New(ctx, traceOpts...) + if err != nil { + return nil, fmt.Errorf("init trace exporter: %w", err) + } + metricExporter, err := otlpmetrichttp.New(ctx, metricOpts...) + if err != nil { + return nil, fmt.Errorf("init metric exporter: %w", err) + } + + env := os.Getenv("GO_ENV") + if env == "" { + env = "dev" + } + res, err := resource.New(ctx, + resource.WithFromEnv(), + resource.WithProcess(), + resource.WithOS(), + resource.WithHost(), + resource.WithAttributes( + semconv.ServiceName("caatsm"), + attribute.String("deployment.environment", env), + ), + ) + if err != nil { + return nil, fmt.Errorf("build telemetry resource: %w", err) + } + + tp := sdktrace.NewTracerProvider( + sdktrace.WithBatcher(traceExporter), + sdktrace.WithResource(res), + ) + mp := sdkmetric.NewMeterProvider( + sdkmetric.WithResource(res), + sdkmetric.WithReader(sdkmetric.NewPeriodicReader(metricExporter)), + ) + + otel.SetTracerProvider(tp) + otel.SetMeterProvider(mp) + otel.SetTextMapPropagator(propagation.TraceContext{}) + + shutdown := func(ctx context.Context) error { + errs := []error{} + if err := mp.Shutdown(ctx); err != nil { + errs = append(errs, err) + } + if err := tp.Shutdown(ctx); err != nil { + errs = append(errs, err) + } + return errors.Join(errs...) + } + + return shutdown, nil +} diff --git a/cmd/main/main_suite_test.go b/cmd/main/main_suite_test.go new file mode 100644 index 0000000..42a2bd9 --- /dev/null +++ b/cmd/main/main_suite_test.go @@ -0,0 +1,13 @@ +package main + +import ( + "testing" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +func TestMainCmd(t *testing.T) { + RegisterFailHandler(Fail) + RunSpecs(t, "cmd/main Suite") +} diff --git a/cmd/main/main_test.go b/cmd/main/main_test.go new file mode 100644 index 0000000..dbab286 --- /dev/null +++ b/cmd/main/main_test.go @@ -0,0 +1,114 @@ +package main + +import ( + "flag" + "time" + + appconfig "caatsm/internal/infra/config" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + "github.com/urfave/cli/v2" +) + +var _ = Describe("CLI overrides", func() { + var ( + cfg *appconfig.Config + flagSet *flag.FlagSet + args []string + ) + + BeforeEach(func() { + cfg = &appconfig.Config{ + NATS: appconfig.NATSConfig{ + URL: "nats://base:4222", + Stream: "BASE_STREAM", + Consumer: "base-consumer", + }, + Subscription: appconfig.SubscriptionConfig{ + Topic: "base.subject", + }, + Publisher: appconfig.PublisherConfig{ + Topic: "base.publisher", + }, + Postgres: appconfig.PostgresConfig{ + URL: "postgres://base", + }, + Log: appconfig.LogConfig{ + Level: "info", + }, + Telemetry: appconfig.TelemetryConfig{ + Enabled: false, + Endpoint: "", + Insecure: false, + }, + } + + flagSet = flag.NewFlagSet("test", flag.ContinueOnError) + flagSet.String("nats-url", "", "") + flagSet.String("subject", "", "") + flagSet.String("stream", "", "") + flagSet.String("consumer", "", "") + flagSet.String("publisher-topic", "", "") + flagSet.String("postgres-url", "", "") + flagSet.String("log-level", "", "") + flagSet.String("replay-from", "", "") + flagSet.Duration("ack-wait", 0, "") + flagSet.Bool("telemetry-enabled", false, "") + flagSet.String("telemetry-endpoint", "", "") + flagSet.Bool("telemetry-insecure", false, "") + + args = []string{ + "--nats-url", "nats://override:4222", + "--subject", "override.subject", + "--stream", "STREAM", + "--consumer", "CONSUMER", + "--publisher-topic", "pub.topic", + "--postgres-url", "postgres://override", + "--log-level", "debug", + "--replay-from", "seq:42", + "--ack-wait", "45s", + "--telemetry-enabled", + "--telemetry-endpoint", "http://otel:4318", + "--telemetry-insecure", + } + }) + + JustBeforeEach(func() { + err := flagSet.Parse(args) + Expect(err).NotTo(HaveOccurred()) + + ctx := cli.NewContext(cli.NewApp(), flagSet, nil) + applyCLIOverrides(cfg, ctx) + }) + + It("applies overrides for all critical flags", func() { + Expect(cfg.NATS.URL).To(Equal("nats://override:4222")) + Expect(cfg.Subscription.Topic).To(Equal("override.subject")) + Expect(cfg.NATS.Stream).To(Equal("STREAM")) + Expect(cfg.NATS.Consumer).To(Equal("CONSUMER")) + Expect(cfg.Publisher.Topic).To(Equal("pub.topic")) + Expect(cfg.Postgres.URL).To(Equal("postgres://override")) + Expect(cfg.Log.Level).To(Equal("debug")) + Expect(cfg.NATS.ConsumerRules.DeliverPolicy).To(Equal("sequence")) + Expect(cfg.NATS.ConsumerRules.StartSequence).To(Equal(uint64(42))) + Expect(cfg.NATS.ConsumerRules.AckWait).To(Equal(45 * time.Second)) + Expect(cfg.Telemetry.Enabled).To(BeTrue()) + Expect(cfg.Telemetry.Endpoint).To(Equal("http://otel:4318")) + Expect(cfg.Telemetry.Insecure).To(BeTrue()) + }) + + Context("with time-based replay override", func() { + BeforeEach(func() { + args = []string{ + "--replay-from", "time:2024-11-15T08:00:00Z", + } + }) + + It("configures deliver policy and start time", func() { + Expect(cfg.NATS.ConsumerRules.DeliverPolicy).To(Equal("time")) + Expect(cfg.NATS.ConsumerRules.StartTime).To(Equal("2024-11-15T08:00:00Z")) + Expect(cfg.NATS.ConsumerRules.StartSequence).To(Equal(uint64(0))) + }) + }) +}) diff --git a/configs/config.dev.toml b/configs/config.dev.toml index 301a15b..e007cf2 100644 --- a/configs/config.dev.toml +++ b/configs/config.dev.toml @@ -17,10 +17,15 @@ replicas = 1 max_deliver = 5 ack_wait = "30s" max_ack_pending = 1024 +deliver_policy = "all" +replay_policy = "instant" +backoff = ["5s", "30s", "2m"] +start_sequence = 0 +start_time = "" [subscription] topic = "telegram.serial" -queue = "tele-queue" +queue_group = "tele-queue" [publisher] topic = "telegram.json" @@ -45,6 +50,7 @@ monitor_interval = "30s" level = "info" format = "json" -[hasura] -endpoint = "http://localhost:8080/v1/graphql" -secret = "aviation-test" \ No newline at end of file +[telemetry] +enabled = false +endpoint = "http://otel-collector:4318" +insecure = true \ No newline at end of file diff --git a/go.mod b/go.mod index ba5e8be..0e2ce23 100644 --- a/go.mod +++ b/go.mod @@ -11,23 +11,32 @@ require ( github.com/knadh/koanf/providers/file v1.2.0 github.com/knadh/koanf/v2 v2.3.0 github.com/nats-io/nats.go v1.47.0 - github.com/onsi/ginkgo/v2 v2.25.1 + github.com/onsi/ginkgo/v2 v2.27.2 github.com/onsi/gomega v1.38.2 - github.com/spf13/viper v1.21.0 github.com/urfave/cli/v2 v2.27.7 + go.opentelemetry.io/otel v1.38.0 + go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetrichttp v1.38.0 + go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.38.0 + go.opentelemetry.io/otel/metric v1.38.0 + go.opentelemetry.io/otel/sdk v1.38.0 + go.opentelemetry.io/otel/sdk/metric v1.38.0 + go.opentelemetry.io/otel/trace v1.38.0 go.uber.org/zap v1.27.0 gopkg.in/natefinch/lumberjack.v2 v2.2.1 ) require ( github.com/Masterminds/semver/v3 v3.4.0 // indirect + github.com/cenkalti/backoff/v5 v5.0.3 // indirect github.com/cpuguy83/go-md2man/v2 v2.0.7 // indirect github.com/fsnotify/fsnotify v1.9.0 // indirect github.com/go-logr/logr v1.4.3 // indirect + github.com/go-logr/stdr v1.2.2 // indirect github.com/go-task/slim-sprig/v3 v3.0.0 // indirect github.com/go-viper/mapstructure/v2 v2.4.0 // indirect github.com/google/go-cmp v0.7.0 // indirect - github.com/google/pprof v0.0.0-20250403155104-27863c87afa6 // indirect + github.com/google/pprof v0.0.0-20251007162407-5df77e3f7d1d // indirect + github.com/grpc-ecosystem/grpc-gateway/v2 v2.27.3 // indirect github.com/jackc/pgpassfile v1.0.0 // indirect github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect github.com/jackc/puddle/v2 v2.2.2 // indirect @@ -38,21 +47,22 @@ require ( github.com/nats-io/nkeys v0.4.11 // indirect github.com/nats-io/nuid v1.0.1 // indirect github.com/pelletier/go-toml v1.9.5 // indirect - github.com/pelletier/go-toml/v2 v2.2.4 // indirect github.com/russross/blackfriday/v2 v2.1.0 // indirect - github.com/sagikazarmark/locafero v0.12.0 // indirect - github.com/spf13/afero v1.15.0 // indirect - github.com/spf13/cast v1.10.0 // indirect - github.com/spf13/pflag v1.0.10 // indirect - github.com/subosito/gotenv v1.6.0 // indirect github.com/xrash/smetrics v0.0.0-20250705151800-55b8f293f342 // indirect - go.uber.org/automaxprocs v1.6.0 // indirect + go.opentelemetry.io/auto/sdk v1.2.1 // indirect + 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/v3 v3.0.4 // indirect golang.org/x/crypto v0.44.0 // indirect + golang.org/x/mod v0.30.0 // indirect golang.org/x/net v0.47.0 // indirect golang.org/x/sync v0.18.0 // indirect golang.org/x/sys v0.38.0 // indirect golang.org/x/text v0.31.0 // indirect golang.org/x/tools v0.39.0 // indirect + google.golang.org/genproto/googleapis/api v0.0.0-20251111163417-95abcf5c77ba // indirect + google.golang.org/genproto/googleapis/rpc v0.0.0-20251111163417-95abcf5c77ba // indirect + google.golang.org/grpc v1.76.0 // indirect + google.golang.org/protobuf v1.36.10 // indirect ) diff --git a/internal/adapter/mapper/suite_test.go b/internal/adapter/mapper/suite_test.go new file mode 100644 index 0000000..03a87e5 --- /dev/null +++ b/internal/adapter/mapper/suite_test.go @@ -0,0 +1,13 @@ +package mapper + +import ( + "testing" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +func TestMapper(t *testing.T) { + RegisterFailHandler(Fail) + RunSpecs(t, "Adapter Mapper Suite") +} diff --git a/internal/adapter/mapper/telegram.go b/internal/adapter/mapper/telegram.go index 132c376..c1e4ba4 100644 --- a/internal/adapter/mapper/telegram.go +++ b/internal/adapter/mapper/telegram.go @@ -43,6 +43,11 @@ func (m *TelegramMapper) ToDBRow(msg *domain.ParsedMessage) ([]interface{}, erro // SecondaryAddresses is already a string, so we can use it directly secondaryAddresses := msg.SecondaryAddresses + status := msg.Status + if status == "" { + status = domain.MessageStatusUnknown + } + return []interface{}{ msgUUID, // uuid msg.MessageID, // message_id @@ -55,6 +60,8 @@ func (m *TelegramMapper) ToDBRow(msg *domain.ParsedMessage) ([]interface{}, erro msg.Category, // category msg.Content, // content (TEXT, original message) bodyDataJSON, // body_data (JSONB) + string(status), // status + msg.ErrorReason, // error_reason msg.ReceivedAt, // received_at msg.ParsedAt, // parsed_at msg.DispatchedAt, // dispatched_at @@ -64,7 +71,7 @@ func (m *TelegramMapper) ToDBRow(msg *domain.ParsedMessage) ([]interface{}, erro // FromDBRow converts a database row to a domain.ParsedMessage func (m *TelegramMapper) FromDBRow(row []interface{}) (*domain.ParsedMessage, error) { - const expectedColumns = 15 + const expectedColumns = 17 if len(row) < expectedColumns { return nil, fmt.Errorf("expected %d columns, got %d", expectedColumns, len(row)) } @@ -94,6 +101,19 @@ func (m *TelegramMapper) FromDBRow(row []interface{}) (*domain.ParsedMessage, er return time.Time{} } + toBool := func(v interface{}) bool { + switch val := v.(type) { + case bool: + return val + case *bool: + return val != nil && *val + case int64: + return val != 0 + default: + return false + } + } + var bodyData interface{} if raw := row[10]; raw != nil { switch val := raw.(type) { @@ -108,7 +128,10 @@ func (m *TelegramMapper) FromDBRow(row []interface{}) (*domain.ParsedMessage, er } } - needDispatch, _ := row[14].(bool) + status := domain.MessageStatusUnknown + if rawStatus := toString(row[11]); rawStatus != "" { + status = domain.MessageStatus(rawStatus) + } return &domain.ParsedMessage{ Uuid: msgUUID.String(), @@ -122,9 +145,11 @@ func (m *TelegramMapper) FromDBRow(row []interface{}) (*domain.ParsedMessage, er Category: toString(row[8]), Content: toString(row[9]), BodyData: bodyData, - ReceivedAt: parseTime(row[11]), - ParsedAt: parseTime(row[12]), - DispatchedAt: parseTime(row[13]), - NeedDispatch: needDispatch, + Status: status, + ErrorReason: toString(row[12]), + ReceivedAt: parseTime(row[13]), + ParsedAt: parseTime(row[14]), + DispatchedAt: parseTime(row[15]), + NeedDispatch: toBool(row[16]), }, nil } diff --git a/internal/adapter/mapper/telegram_test.go b/internal/adapter/mapper/telegram_test.go index b32bd95..3b299f5 100644 --- a/internal/adapter/mapper/telegram_test.go +++ b/internal/adapter/mapper/telegram_test.go @@ -1,71 +1,68 @@ package mapper import ( - "testing" "time" "caatsm/internal/domain" "github.com/google/uuid" + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" ) -func TestTelegramMapper_ToDBRow_GeneratesUUIDWhenEmpty(t *testing.T) { - mapper := NewTelegramMapper() - msg := &domain.ParsedMessage{} +var _ = Describe("TelegramMapper", func() { + var mapper *TelegramMapper - row, err := mapper.ToDBRow(msg) - if err != nil { - t.Fatalf("unexpected error: %v", err) - } + BeforeEach(func() { + mapper = NewTelegramMapper() + }) - value, ok := row[0].(uuid.UUID) - if !ok { - t.Fatalf("expected first column to be uuid.UUID, got %T", row[0]) - } - if value == uuid.Nil { - t.Fatalf("expected generated uuid to be non-nil") - } -} + Describe("ToDBRow", func() { + It("generates a UUID when missing", func() { + msg := &domain.ParsedMessage{} -func TestTelegramMapper_FromDBRow_RoundTrip(t *testing.T) { - mapper := NewTelegramMapper() - now := time.Now().UTC() + row, err := mapper.ToDBRow(msg) + Expect(err).NotTo(HaveOccurred()) - original := &domain.ParsedMessage{ - Uuid: uuid.NewString(), - MessageID: "TMQ1324", - DateTime: "150631", - PriorityIndicator: "FF", - PrimaryAddress: "ZBTJZPZX", - SecondaryAddresses: "150630 ZBACZQZX", - Originator: "ORIGIN", - OriginatorDateTime: "150630", - Category: "FPL", - Content: "raw telegram", - BodyData: map[string]string{"key": "value"}, - ReceivedAt: now, - ParsedAt: now, - DispatchedAt: now, - NeedDispatch: true, - } + value, ok := row[0].(uuid.UUID) + Expect(ok).To(BeTrue()) + Expect(value).NotTo(Equal(uuid.Nil)) + }) + }) - row, err := mapper.ToDBRow(original) - if err != nil { - t.Fatalf("unexpected error: %v", err) - } + Describe("FromDBRow", func() { + It("round-trips telegram data", func() { + now := time.Now().UTC() + original := &domain.ParsedMessage{ + Uuid: uuid.NewString(), + MessageID: "TMQ1324", + DateTime: "150631", + PriorityIndicator: "FF", + PrimaryAddress: "ZBTJZPZX", + SecondaryAddresses: "150630 ZBACZQZX", + Originator: "ORIGIN", + OriginatorDateTime: "150630", + Category: "FPL", + Content: "raw telegram", + BodyData: map[string]string{"key": "value"}, + ReceivedAt: now, + ParsedAt: now, + DispatchedAt: now, + NeedDispatch: true, + Status: domain.MessageStatusParsed, + } - roundTrip, err := mapper.FromDBRow(row) - if err != nil { - t.Fatalf("unexpected error reading row: %v", err) - } + row, err := mapper.ToDBRow(original) + Expect(err).NotTo(HaveOccurred()) + Expect(row).To(HaveLen(17)) - if roundTrip.Uuid != original.Uuid { - t.Fatalf("expected uuid %s, got %s", original.Uuid, roundTrip.Uuid) - } - if roundTrip.MessageID != original.MessageID { - t.Fatalf("expected message_id %s, got %s", original.MessageID, roundTrip.MessageID) - } - if roundTrip.NeedDispatch != original.NeedDispatch { - t.Fatalf("expected need_dispatch %v, got %v", original.NeedDispatch, roundTrip.NeedDispatch) - } -} + roundTrip, err := mapper.FromDBRow(row) + Expect(err).NotTo(HaveOccurred()) + + Expect(roundTrip.Uuid).To(Equal(original.Uuid)) + Expect(roundTrip.MessageID).To(Equal(original.MessageID)) + Expect(roundTrip.NeedDispatch).To(Equal(original.NeedDispatch)) + Expect(roundTrip.Status).To(Equal(original.Status)) + }) + }) +}) diff --git a/internal/adapter/parser/aviation_parser.go b/internal/adapter/parser/aviation_parser.go index 74a33b7..220612f 100644 --- a/internal/adapter/parser/aviation_parser.go +++ b/internal/adapter/parser/aviation_parser.go @@ -14,8 +14,7 @@ func NewAviationParser() *AviationParser { } // Parse parses a raw message string and returns a ParsedMessage -func (p *AviationParser) Parse(rawText string) *domain.ParsedMessage { +func (p *AviationParser) Parse(rawText string) (*domain.ParsedMessage, 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 b091ce0..dfaec04 100644 --- a/internal/adapter/parser/parser.go +++ b/internal/adapter/parser/parser.go @@ -5,7 +5,5 @@ import "caatsm/internal/domain" // 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 - - // TODO: consider returning (*domain.ParsedMessage, error) to surface parse failures explicitly. + Parse(rawText string) (*domain.ParsedMessage, error) } diff --git a/internal/adapter/repository.go b/internal/adapter/repository.go index 5ca7069..d73d708 100644 --- a/internal/adapter/repository.go +++ b/internal/adapter/repository.go @@ -12,5 +12,8 @@ type Repository interface { // InsertBatch inserts multiple telegram messages in a batch InsertBatch(ctx context.Context, msgs []*domain.ParsedMessage) error + + // InsertRaw captures an unparsed or failed telegram for later analysis. + InsertRaw(ctx context.Context, msg *domain.ParsedMessage) error } diff --git a/internal/app/errors_test.go b/internal/app/errors_test.go index 6650529..91ab9b5 100644 --- a/internal/app/errors_test.go +++ b/internal/app/errors_test.go @@ -2,32 +2,24 @@ package app import ( "errors" - "testing" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" ) -func TestPermanentWrapsError(t *testing.T) { - base := errors.New("boom") - perr := Permanent(base) +var _ = Describe("Permanent errors", func() { + It("wraps errors and reports permanence", func() { + base := errors.New("boom") + perr := Permanent(base) - if perr == nil { - t.Fatalf("expected wrapped error, got nil") - } - if !IsPermanent(perr) { - t.Fatalf("expected IsPermanent to be true") - } - if !errors.Is(perr, base) { - t.Fatalf("expected wrapped error to unwrap to base") - } - if errors.Is(base, perr) { - t.Fatalf("expected base not to consider wrapper as same") - } -} + Expect(perr).NotTo(BeNil()) + Expect(IsPermanent(perr)).To(BeTrue()) + Expect(errors.Is(perr, base)).To(BeTrue()) + Expect(errors.Is(base, perr)).To(BeFalse()) + }) -func TestPermanentNil(t *testing.T) { - if Permanent(nil) != nil { - t.Fatalf("Permanent(nil) should return nil") - } - if IsPermanent(nil) { - t.Fatalf("IsPermanent(nil) should be false") - } -} + It("treats nil as non-permanent", func() { + Expect(Permanent(nil)).To(BeNil()) + Expect(IsPermanent(nil)).To(BeFalse()) + }) +}) diff --git a/internal/app/processor.go b/internal/app/processor.go index ec79e34..2b1f41a 100644 --- a/internal/app/processor.go +++ b/internal/app/processor.go @@ -3,11 +3,16 @@ package app import ( "caatsm/internal/adapter" "caatsm/internal/adapter/parser" + "caatsm/internal/domain" "context" "fmt" "strings" "time" + "go.opentelemetry.io/otel" + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/codes" + "go.opentelemetry.io/otel/trace" "go.uber.org/zap" ) @@ -40,11 +45,20 @@ func (p *MessageProcessor) Handle(ctx context.Context, raw []byte, msgID string) return Permanent(fmt.Errorf("empty message")) } + tracer := otel.Tracer("caatsm/app") + ctx, span := tracer.Start(ctx, "MessageProcessor.Handle") + defer span.End() + span.SetAttributes(attribute.String("nats.msg_id", msgID)) + receivedAt := time.Now() - parsed := p.parser.Parse(string(raw)) + parsed, parseErr := p.parser.Parse(string(raw)) if parsed == nil { - return Permanent(fmt.Errorf("parser returned nil")) + parsed = domain.NewParsedMessage() + parsed.Content = string(raw) + parsed.ErrorReason = "parser returned nil" + parsed.Status = domain.MessageStatusBodyError + parseErr = fmt.Errorf("parser returned nil") } if msgID != "" { @@ -60,44 +74,102 @@ 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 parseErr == nil { + parsed.Status = domain.MessageStatusParsed + } else { + parsed.Status = domain.MessageStatusBodyError + } + } + + if parseErr != nil || !parsed.Parsed { + if parsed.ErrorReason == "" && parseErr != nil { + parsed.ErrorReason = parseErr.Error() + } + span.RecordError(parseErr) + span.SetStatus(codes.Error, parseErr.Error()) + p.persistRaw(ctx, parsed) + p.logger.Warn("Message failed to parse", + zap.String("msg_id", msgID), + zap.String("status", string(parsed.Status)), + zap.String("content_preview", truncateContent(parsed.Content, 256)), + zap.Error(parseErr), + ) + return Permanent(fmt.Errorf("parser error: %w", parseErr)) + } + parsed.ErrorReason = "" // Log parsing result - if !parsed.Parsed { - p.logger.Warn("Message not parsed", - zap.String("msg_id", msgID), - zap.String("message_id", parsed.MessageID), - zap.String("category", parsed.Category), - zap.String("content_preview", truncateContent(parsed.Content, 256)), - ) - } else { - p.logger.Info("Message parsed successfully", - zap.String("msg_id", msgID), - zap.String("message_id", parsed.MessageID), - zap.String("category", parsed.Category), - zap.Time("received_at", parsed.ReceivedAt), - zap.Time("parsed_at", parsed.ParsedAt), - ) - } + span.SetAttributes( + attribute.String("telegram.status", string(parsed.Status)), + attribute.Bool("telegram.parsed", parsed.Parsed), + attribute.String("telegram.category", parsed.Category), + ) + + p.logger.Info("Message parsed successfully", + zap.String("msg_id", msgID), + zap.String("message_id", parsed.MessageID), + zap.String("category", parsed.Category), + zap.Time("received_at", parsed.ReceivedAt), + zap.Time("parsed_at", parsed.ParsedAt), + ) // Insert into database if err := p.repository.InsertOne(ctx, parsed); err != nil { + span.RecordError(err) + span.SetStatus(codes.Error, err.Error()) return fmt.Errorf("failed to insert message: %w", err) } // Publish parsed message + _, pubSpan := tracer.Start(ctx, "Publisher.Publish") if err := p.publisher.Publish(parsed); err != nil { // Log error but don't fail the entire operation p.logger.Error("Failed to publish message", zap.String("msg_id", msgID), zap.Error(err), ) + pubSpan.RecordError(err) + pubSpan.SetStatus(codes.Error, err.Error()) + parsed.Status = domain.MessageStatusPublishFail + parsed.ErrorReason = err.Error() + p.persistRaw(ctx, parsed) // Mark as permanent so the consumer will ack instead of retrying + pubSpan.End() return Permanent(fmt.Errorf("failed to publish message: %w", err)) } + pubSpan.End() return nil } +func (p *MessageProcessor) persistRaw(ctx context.Context, msg *domain.ParsedMessage) { + if msg == nil || p.repository == nil { + return + } + if msg.Content == "" && msg.BodyData != nil { + msg.Content = fmt.Sprintf("%v", msg.BodyData) + } + if msg.ReceivedAt.IsZero() { + msg.ReceivedAt = time.Now() + } + if err := p.repository.InsertRaw(ctx, msg); err != nil { + p.logger.Error("Failed to persist raw telegram", + zap.String("message_id", msg.MessageID), + zap.String("status", string(msg.Status)), + zap.Error(err), + ) + } else { + if span := trace.SpanFromContext(ctx); span.IsRecording() { + span.AddEvent("raw telegram persisted", + trace.WithAttributes( + attribute.String("telegram.status", string(msg.Status)), + attribute.String("telegram.message_id", msg.MessageID), + )) + } + } +} + func truncateContent(content string, limit int) string { if limit <= 0 || len(content) <= limit { return content diff --git a/internal/app/processor_test.go b/internal/app/processor_test.go index 40b29ad..f4f59a4 100644 --- a/internal/app/processor_test.go +++ b/internal/app/processor_test.go @@ -4,7 +4,6 @@ import ( "context" "errors" "strings" - "testing" "time" "caatsm/internal/adapter" @@ -12,166 +11,137 @@ import ( "caatsm/internal/domain" "github.com/google/uuid" + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" "go.uber.org/zap" "go.uber.org/zap/zaptest/observer" ) -func TestHandleEmptyMessageIsPermanent(t *testing.T) { - proc := newTestProcessor(&stubParser{}, &stubRepository{}, &stubPublisher{}) - err := proc.Handle(context.Background(), nil, "id-1") - if err == nil || !IsPermanent(err) { - t.Fatalf("expected permanent error for empty message, got %v", err) - } -} +var _ = Describe("MessageProcessor", func() { + var ( + repo *stubRepository + pub *stubPublisher + proc *MessageProcessor + ctx context.Context + parserStub *stubParser + ) -func TestHandleNilParserResultIsPermanent(t *testing.T) { - proc := newTestProcessor(&stubParser{value: nil}, &stubRepository{}, &stubPublisher{}) - err := proc.Handle(context.Background(), []byte("payload"), "id-2") - if err == nil || !IsPermanent(err) { - t.Fatalf("expected permanent error for nil parser result, got %v", err) - } -} + BeforeEach(func() { + repo = &stubRepository{} + pub = &stubPublisher{} + parserStub = &stubParser{} + proc = newTestProcessor(parserStub, repo, pub) + ctx = context.Background() + }) -func TestHandleSuccessDoesNotOverwriteUuid(t *testing.T) { - originalUUID := uuid.NewString() - parsed := &domain.ParsedMessage{Uuid: originalUUID, Parsed: true} + Describe("Handle", func() { + It("returns a permanent error when payload is empty", func() { + err := proc.Handle(ctx, nil, "id-1") + Expect(err).To(HaveOccurred()) + Expect(IsPermanent(err)).To(BeTrue()) + }) - repo := &stubRepository{} - pub := &stubPublisher{} - proc := newTestProcessor(&stubParser{value: parsed}, repo, pub) + It("records raw messages when parser returns nil", func() { + parserStub.value = nil + err := proc.Handle(ctx, []byte("payload"), "id-2") + Expect(err).To(HaveOccurred()) + Expect(IsPermanent(err)).To(BeTrue()) + Expect(repo.rawCount()).To(Equal(1)) + }) - const msgID = "msg-123" - if err := proc.Handle(context.Background(), []byte("payload"), msgID); err != nil { - t.Fatalf("expected success, got %v", err) - } + It("preserves UUIDs and appends nats message id comment", func() { + originalUUID := uuid.NewString() + parserStub.value = &domain.ParsedMessage{Uuid: originalUUID, Parsed: true, Status: domain.MessageStatusParsed} - if repo.last() == nil { - t.Fatalf("expected message to be inserted") - } - if repo.last().Uuid != originalUUID { - t.Fatalf("expected uuid to remain %s, got %s", originalUUID, repo.last().Uuid) - } - if !strings.Contains(repo.last().Comments, "nats_msg_id=msg-123") { - t.Fatalf("expected comments to contain msg id, got %q", repo.last().Comments) - } - if pub.last == nil { - t.Fatalf("expected publisher to receive message") - } -} + Expect(proc.Handle(ctx, []byte("payload"), "msg-123")).To(Succeed()) -func TestHandlePublisherErrorIsPermanent(t *testing.T) { - parsed := &domain.ParsedMessage{Parsed: true} + Expect(repo.last()).NotTo(BeNil()) + Expect(repo.last().Uuid).To(Equal(originalUUID)) + Expect(repo.last().Comments).To(ContainSubstring("nats_msg_id=msg-123")) + Expect(pub.last).NotTo(BeNil()) + }) - repo := &stubRepository{} - pub := &stubPublisher{err: errors.New("publish failed")} - proc := newTestProcessor(&stubParser{value: parsed}, repo, pub) + It("treats publisher failures as permanent and stores raw entries", func() { + parserStub.value = &domain.ParsedMessage{Parsed: true, Status: domain.MessageStatusParsed} + pub.err = errors.New("publish failed") - err := proc.Handle(context.Background(), []byte("payload"), "id-3") - if err == nil { - t.Fatalf("expected error when publisher fails") - } - if !IsPermanent(err) { - t.Fatalf("publisher failure should be permanent") - } - if repo.last() == nil { - t.Fatalf("expected message to insert before publish failure") - } -} + err := proc.Handle(ctx, []byte("payload"), "id-3") + Expect(err).To(HaveOccurred()) + Expect(IsPermanent(err)).To(BeTrue()) + Expect(repo.last()).NotTo(BeNil()) + Expect(repo.rawCount()).To(Equal(1)) + Expect(repo.lastRaw().Status).To(Equal(domain.MessageStatusPublishFail)) + }) -func TestMessageProcessor_Handle_SetsReceivedAndParsedAtWhenZero(t *testing.T) { - parsed := &domain.ParsedMessage{ - Uuid: uuid.NewString(), - Parsed: true, - } - repo := &stubRepository{} - publisher := &stubPublisher{} - logger := zap.NewNop() - processor := NewMessageProcessor(&stubParser{value: parsed}, repo, publisher, logger) + It("sets timestamps when missing", func() { + parserStub.value = &domain.ParsedMessage{ + Uuid: uuid.NewString(), + Parsed: true, + Status: domain.MessageStatusParsed, + } + pub.err = nil - start := time.Now() - if err := processor.Handle(context.Background(), []byte("raw"), "msg-4"); err != nil { - t.Fatalf("unexpected error: %v", err) - } - saved := repo.last() - if saved == nil { - t.Fatal("expected repository to receive a message") - } - if saved.ReceivedAt.IsZero() || saved.ParsedAt.IsZero() { - t.Fatalf("expected timestamps to be set, got received=%v parsed=%v", saved.ReceivedAt, saved.ParsedAt) - } - if saved.ReceivedAt.Before(start.Add(-time.Second)) || saved.ParsedAt.Before(start.Add(-time.Second)) { - t.Fatalf("timestamps look stale: received=%v parsed=%v", saved.ReceivedAt, saved.ParsedAt) - } -} + start := time.Now() + Expect(proc.Handle(ctx, []byte("payload"), "msg-4")).To(Succeed()) -func TestMessageProcessor_Handle_DoesNotOverrideExistingTimestamps(t *testing.T) { - received := time.Now().Add(-2 * time.Minute) - parsedAt := time.Now().Add(-time.Minute) + saved := repo.last() + Expect(saved).NotTo(BeNil()) + Expect(saved.ReceivedAt).NotTo(BeZero()) + Expect(saved.ParsedAt).NotTo(BeZero()) + Expect(saved.ReceivedAt.After(start.Add(-time.Second))).To(BeTrue()) + Expect(saved.ParsedAt.After(start.Add(-time.Second))).To(BeTrue()) + }) - parsed := &domain.ParsedMessage{ - Uuid: uuid.NewString(), - Parsed: true, - ReceivedAt: received, - ParsedAt: parsedAt, - } - repo := &stubRepository{} - publisher := &stubPublisher{} - logger := zap.NewNop() - processor := NewMessageProcessor(&stubParser{value: parsed}, repo, publisher, logger) + 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{ + Uuid: uuid.NewString(), + Parsed: true, + Status: domain.MessageStatusParsed, + ReceivedAt: received, + ParsedAt: parsedAt, + } - if err := processor.Handle(context.Background(), []byte("raw"), "msg-5"); err != nil { - t.Fatalf("unexpected error: %v", err) - } - saved := repo.last() - if saved.ReceivedAt != received { - t.Fatalf("expected received_at to remain %v, got %v", received, saved.ReceivedAt) - } - if saved.ParsedAt != parsedAt { - t.Fatalf("expected parsed_at to remain %v, got %v", parsedAt, saved.ParsedAt) - } -} + Expect(proc.Handle(ctx, []byte("payload"), "msg-5")).To(Succeed()) + Expect(repo.last().ReceivedAt).To(Equal(received)) + Expect(repo.last().ParsedAt).To(Equal(parsedAt)) + }) -func TestContentPreview_TruncatesLongContent(t *testing.T) { - longContent := strings.Repeat("a", 1024) - preview := truncateContent(longContent, 256) - if len(preview) != 256 { - t.Fatalf("expected preview length 256, got %d", len(preview)) - } - if !strings.HasSuffix(preview, "...") { - t.Fatalf("expected preview to end with ellipsis, got %q", preview[len(preview)-10:]) - } -} + It("logs truncated previews when parsing fails", func() { + core, logs := observer.New(zap.WarnLevel) + logger := zap.New(core) + parserStub = &stubParser{ + value: &domain.ParsedMessage{ + Content: strings.Repeat("x", 1024), + Parsed: false, + Status: domain.MessageStatusBodyError, + ErrorReason: "parse failure", + }, + err: errors.New("parse failure"), + } + proc = NewMessageProcessor(parserStub, repo, pub, logger) -func TestMessageProcessor_Handle_NotParsedLogsPreviewOnly(t *testing.T) { - core, logs := observer.New(zap.WarnLevel) - logger := zap.New(core) + err := proc.Handle(ctx, []byte("raw"), "msg-6") + Expect(err).To(HaveOccurred()) + Expect(IsPermanent(err)).To(BeTrue()) - parser := &stubParser{ - value: &domain.ParsedMessage{ - Content: strings.Repeat("x", 1024), - Parsed: false, - }, - } - repo := &stubRepository{} - publisher := &stubPublisher{} - processor := NewMessageProcessor(parser, repo, publisher, logger) + entries := logs.FilterMessage("Message failed to parse").All() + Expect(entries).NotTo(BeEmpty()) + preview, ok := entries[0].ContextMap()["content_preview"].(string) + Expect(ok).To(BeTrue()) + Expect(len(preview)).To(BeNumerically("<=", 256)) + }) + }) - if err := processor.Handle(context.Background(), []byte("raw"), "msg-6"); err != nil { - t.Fatalf("unexpected error: %v", err) - } - - entries := logs.FilterMessage("Message not parsed").All() - if len(entries) == 0 { - t.Fatal("expected a warning log for unparsed message") - } - preview, ok := entries[0].ContextMap()["content_preview"].(string) - if !ok { - t.Fatal("expected content_preview field in log") - } - if len(preview) > 256 { - t.Fatalf("expected preview <= 256 chars, got %d", len(preview)) - } -} + Describe("truncateContent", func() { + It("keeps length at limit with ellipsis", func() { + longContent := strings.Repeat("a", 1024) + Expect(truncateContent(longContent, 256)).To(HaveLen(256)) + Expect(truncateContent(longContent, 256)).To(HaveSuffix("...")) + }) + }) +}) func newTestProcessor(p parser.Parser, repo adapter.Repository, pub adapter.Publisher) *MessageProcessor { return NewMessageProcessor(p, repo, pub, zap.NewNop()) @@ -179,15 +149,18 @@ func newTestProcessor(p parser.Parser, repo adapter.Repository, pub adapter.Publ type stubParser struct { value *domain.ParsedMessage + err error } -func (s *stubParser) Parse(rawText string) *domain.ParsedMessage { - return s.value +func (s *stubParser) Parse(rawText string) (*domain.ParsedMessage, error) { + return s.value, s.err } type stubRepository struct { inserted []*domain.ParsedMessage + raw []*domain.ParsedMessage err error + rawErr error } func (s *stubRepository) InsertOne(ctx context.Context, msg *domain.ParsedMessage) error { @@ -202,6 +175,14 @@ func (s *stubRepository) InsertBatch(ctx context.Context, msgs []*domain.ParsedM return errors.New("not implemented") } +func (s *stubRepository) InsertRaw(ctx context.Context, msg *domain.ParsedMessage) error { + if s.rawErr != nil { + return s.rawErr + } + s.raw = append(s.raw, msg) + return nil +} + func (s *stubRepository) last() *domain.ParsedMessage { if len(s.inserted) == 0 { return nil @@ -209,6 +190,17 @@ func (s *stubRepository) last() *domain.ParsedMessage { return s.inserted[len(s.inserted)-1] } +func (s *stubRepository) lastRaw() *domain.ParsedMessage { + if len(s.raw) == 0 { + return nil + } + return s.raw[len(s.raw)-1] +} + +func (s *stubRepository) rawCount() int { + return len(s.raw) +} + type stubPublisher struct { last interface{} err error diff --git a/internal/app/suite_test.go b/internal/app/suite_test.go new file mode 100644 index 0000000..86f6717 --- /dev/null +++ b/internal/app/suite_test.go @@ -0,0 +1,13 @@ +package app + +import ( + "testing" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +func TestApp(t *testing.T) { + RegisterFailHandler(Fail) + RunSpecs(t, "App Suite") +} diff --git a/internal/config/config.go b/internal/config/config.go deleted file mode 100644 index fbef9ee..0000000 --- a/internal/config/config.go +++ /dev/null @@ -1,124 +0,0 @@ -package config - -import ( - "fmt" - "os" - "regexp" - "strings" - "time" - - "github.com/spf13/viper" -) - -var MyConfig *Config - -type Config struct { - Nats NatsConfig - Subscription SubscriptionConfig - Publisher PublisherConfig - Timeouts TimeoutsConfig - Hasura HasuraConfig -} - -type NatsConfig struct { - Client string - URL string - Cluster string -} - -type SubscriptionConfig struct { - Topic string `mapstructure:"topic"` - QueueGroup string `mapstructure:"queue_group"` -} - -type PublisherConfig struct { - Topic string `mapstructure:"topic"` -} - -type TimeoutsConfig struct { - Server time.Duration `mapstructure:"server"` - ReconnectWait time.Duration `mapstructure:"reconnect_wait"` - Close time.Duration `mapstructure:"close"` - AckWait time.Duration `mapstructure:"ack_wait"` -} - -type BodyConfig struct { - Patterns []PatternConfig -} - -type PatternConfig struct { - Pattern string - Comments string - Expression *regexp.Regexp -} - -type HasuraConfig struct { - Endpoint string - Secret string -} - -const ( - EnvProd = "prod" - EnvDev = "dev" - EnvTest = "test" -) - -func SetMyConfig(cfg *Config) { - MyConfig = cfg -} - -func GetMyConfig() *Config { - if MyConfig == nil { - cfg, err := LoadConfig() - if err != nil { - fmt.Printf("error loading config: %v", err) - } - MyConfig = cfg - } - return MyConfig -} - -// LoadConfig loads the configuration from a file -func LoadConfig() (*Config, error) { - // log := utils.Logger - env := os.Getenv("GO_ENV") - if env == "" { - env = "dev" - } - // log.Infof("Environment: %s", env) - - viper.SetConfigType("toml") - viper.SetConfigName("config." + env) - viper.AddConfigPath("configs") - viper.SetEnvPrefix("tele") - viper.SetEnvKeyReplacer(strings.NewReplacer(".", "_")) - - if err := viper.ReadInConfig(); err != nil { - // log.Errorf("error reading config file for environment '%s': %v", env, err) - return nil, fmt.Errorf("error reading config file for environment '%s': %w", env, err) - } - - var config Config - if err := viper.Unmarshal(&config); err != nil { - // log.Errorf("unable to decode config into struct for environment '%s': %v", env, err) - return nil, fmt.Errorf("unable to decode config into struct for environment '%s': %w", env, err) - } - return &config, nil -} - -// ValidateConfig validates the loaded configuration -func ValidateConfig(cfg *Config) error { - // log := utils.Logger - - if cfg.Nats.Client == "" { - return fmt.Errorf("nats client is required") - } - if cfg.Nats.URL == "" { - return fmt.Errorf("nats URL is required") - } - if cfg.Subscription.Topic == "" { - return fmt.Errorf("subscription topic is required") - } - // fmt.Println("config validation passed") - return nil -} diff --git a/internal/config/config_test.go b/internal/config/config_test.go deleted file mode 100644 index 08af0ec..0000000 --- a/internal/config/config_test.go +++ /dev/null @@ -1,112 +0,0 @@ -package config - -import ( - "os" - "testing" - - . "github.com/onsi/ginkgo/v2" - . "github.com/onsi/gomega" - "github.com/spf13/viper" -) - -func TestConfig(t *testing.T) { - RegisterFailHandler(Fail) - RunSpecs(t, "Config Suite") -} - -var _ = Describe("Config", func() { - var originalEnv string - - BeforeEach(func() { - // Save the original GO_ENV value - originalEnv = os.Getenv("GO_ENV") - // Set up a temporary configuration file for testing - viper.Reset() - viper.SetConfigType("toml") - configContent := ` -[nats] -client = "test-client" -url = "nats://localhost:4222" -cluster = "test-cluster" - -[subscription] -topic = "example-topic" -queue_group = "example-group" - -[timeouts] -server_timeout = "30s" -reconnect_wait = "10s" -close_timeout = "10s" -ack_wait_timeout = "5s" - -[hasura] -endpoint = "http://localhost:8080/v1/graphql" -secret = "aviation-test" -` - tmpFile, err := os.CreateTemp("", "config.*.toml") - Expect(err).NotTo(HaveOccurred()) - _, err = tmpFile.Write([]byte(configContent)) - Expect(err).NotTo(HaveOccurred()) - err = tmpFile.Close() - Expect(err).NotTo(HaveOccurred()) - - viper.SetConfigFile(tmpFile.Name()) - err = viper.ReadInConfig() - Expect(err).NotTo(HaveOccurred()) - - // Load the configuration - MyConfig = &Config{} - err = viper.Unmarshal(MyConfig) - Expect(err).NotTo(HaveOccurred()) - }) - - AfterEach(func() { - // Restore the original GO_ENV value - os.Setenv("GO_ENV", originalEnv) - }) - - Context("Loading configuration", func() { - It("should load the configuration correctly", func() { - cfg := GetMyConfig() - Expect(cfg).NotTo(BeNil()) - Expect(cfg.Nats.Client).To(Equal("test-client")) - Expect(cfg.Nats.URL).To(Equal("nats://localhost:4222")) - Expect(cfg.Subscription.Topic).To(Equal("example-topic")) - Expect(cfg.Subscription.QueueGroup).To(Equal("example-group")) - Expect(cfg.Hasura.Endpoint).To(Equal("http://localhost:8080/v1/graphql")) - Expect(cfg.Hasura.Secret).To(Equal("aviation-test")) - }) - }) - - Context("Validating configuration", func() { - It("should validate a valid configuration", func() { - cfg := GetMyConfig() - err := ValidateConfig(cfg) - Expect(err).NotTo(HaveOccurred()) - }) - - It("should return an error for missing NATS client", func() { - cfg := GetMyConfig() - cfg.Nats.Client = "" - err := ValidateConfig(cfg) - Expect(err).To(HaveOccurred()) - Expect(err.Error()).To(Equal("nats client is required")) - }) - - It("should return an error for missing NATS URL", func() { - cfg := GetMyConfig() - cfg.Nats.URL = "" - err := ValidateConfig(cfg) - Expect(err).To(HaveOccurred()) - Expect(err.Error()).To(Equal("nats URL is required")) - }) - - It("should return an error for missing subscription topic", func() { - cfg := GetMyConfig() - cfg.Subscription.Topic = "" - err := ValidateConfig(cfg) - Expect(err).To(HaveOccurred()) - Expect(err.Error()).To(Equal("subscription topic is required")) - }) - }) -}) diff --git a/internal/domain/aviation.go b/internal/domain/aviation.go index d77d2d6..ad5aef4 100644 --- a/internal/domain/aviation.go +++ b/internal/domain/aviation.go @@ -84,6 +84,18 @@ 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'). @@ -105,7 +117,8 @@ type ParsedMessage struct { 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 @@ -113,6 +126,7 @@ func NewParsedMessage() *ParsedMessage { return &ParsedMessage{ // SecondaryAddresses: []string{}, Parsed: false, + Status: MessageStatusUnknown, } } diff --git a/internal/iface/interface.go b/internal/iface/interface.go index 9bd48f3..97264b2 100644 --- a/internal/iface/interface.go +++ b/internal/iface/interface.go @@ -1,8 +1,8 @@ package iface import ( - "caatsm/internal/config" "caatsm/internal/domain" + "caatsm/internal/infra/config" ) type MessageHandler interface { diff --git a/internal/infra/config/config.go b/internal/infra/config/config.go index 79def3d..c995fc8 100644 --- a/internal/infra/config/config.go +++ b/internal/infra/config/config.go @@ -19,6 +19,7 @@ type Config struct { App AppConfig `koanf:"app"` Log LogConfig `koanf:"log"` Publisher PublisherConfig `koanf:"publisher"` + Telemetry TelemetryConfig `koanf:"telemetry"` // Legacy fields for backward compatibility during migration Subscription SubscriptionConfig `koanf:"subscription"` Timeouts TimeoutsConfig `koanf:"timeouts"` @@ -51,6 +52,11 @@ type ConsumerRulesConfig struct { MaxDeliver int `koanf:"max_deliver"` AckWait time.Duration `koanf:"ack_wait"` MaxAckPending int `koanf:"max_ack_pending"` + DeliverPolicy string `koanf:"deliver_policy"` + ReplayPolicy string `koanf:"replay_policy"` + Backoff []time.Duration `koanf:"backoff"` + StartSequence uint64 `koanf:"start_sequence"` + StartTime string `koanf:"start_time"` } // PostgresConfig holds PostgreSQL configuration @@ -78,6 +84,13 @@ type PublisherConfig struct { Topic string `koanf:"topic"` } +// TelemetryConfig controls tracing/metrics exporters. +type TelemetryConfig struct { + Enabled bool `koanf:"enabled"` + Endpoint string `koanf:"endpoint"` + Insecure bool `koanf:"insecure"` +} + // SubscriptionConfig holds subscription configuration (legacy) type SubscriptionConfig struct { Topic string `koanf:"topic"` @@ -183,6 +196,15 @@ func LoadConfig() (*Config, error) { if cfg.NATS.ConsumerRules.MaxAckPending == 0 { cfg.NATS.ConsumerRules.MaxAckPending = 1024 } + if cfg.NATS.ConsumerRules.DeliverPolicy == "" { + cfg.NATS.ConsumerRules.DeliverPolicy = "all" + } + if cfg.NATS.ConsumerRules.ReplayPolicy == "" { + cfg.NATS.ConsumerRules.ReplayPolicy = "instant" + } + if cfg.Telemetry.Endpoint == "" { + cfg.Telemetry.Endpoint = "" + } // Validate configuration if err := cfg.Validate(); err != nil { @@ -236,6 +258,29 @@ func (c *Config) Validate() error { if c.NATS.ConsumerRules.MaxAckPending < 0 { return fmt.Errorf("nats.consumer.max_ack_pending must be >= 0") } + switch strings.ToLower(c.NATS.ConsumerRules.DeliverPolicy) { + case "", "all", "new", "last", "last_per_subject", "sequence", "time": + default: + return fmt.Errorf("nats.consumer.deliver_policy must be one of all,new,last,last_per_subject,sequence,time") + } + switch strings.ToLower(c.NATS.ConsumerRules.ReplayPolicy) { + case "", "instant", "original": + default: + return fmt.Errorf("nats.consumer.replay_policy must be instant or original") + } + if c.NATS.ConsumerRules.StartTime != "" { + if _, err := time.Parse(time.RFC3339, c.NATS.ConsumerRules.StartTime); err != nil { + return fmt.Errorf("nats.consumer.start_time must be RFC3339: %w", err) + } + } + for _, d := range c.NATS.ConsumerRules.Backoff { + if d < 0 { + return fmt.Errorf("nats.consumer.backoff durations must be >= 0") + } + } + if c.Telemetry.Endpoint == "" && c.Telemetry.Enabled { + return fmt.Errorf("telemetry.endpoint is required when telemetry.enabled=true") + } return nil } diff --git a/internal/infra/config/config_test.go b/internal/infra/config/config_test.go index 8ee7c26..4d09182 100644 --- a/internal/infra/config/config_test.go +++ b/internal/infra/config/config_test.go @@ -3,35 +3,38 @@ package config import ( "os" "path/filepath" - "testing" "time" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" ) -func TestLoadConfig_DefaultAckWait(t *testing.T) { - t.Setenv("GO_ENV", "testdefaults") +var _ = Describe("LoadConfig", func() { + var ( + originalWD string + ) - wd, err := os.Getwd() - if err != nil { - t.Fatalf("failed to get working dir: %v", err) - } - repoRoot := filepath.Clean(filepath.Join(wd, "..", "..", "..")) - if err := os.Chdir(repoRoot); err != nil { - t.Fatalf("failed to chdir to repo root: %v", err) - } - t.Cleanup(func() { - _ = os.Chdir(wd) + BeforeEach(func() { + Expect(os.Setenv("GO_ENV", "testdefaults")).To(Succeed()) + + var err error + originalWD, err = os.Getwd() + Expect(err).NotTo(HaveOccurred()) + + repoRoot := filepath.Clean(filepath.Join(originalWD, "..", "..", "..")) + Expect(os.Chdir(repoRoot)).To(Succeed()) }) - cfg, err := LoadConfig() - if err != nil { - t.Fatalf("failed to load config: %v", err) - } + AfterEach(func() { + Expect(os.Chdir(originalWD)).To(Succeed()) + }) - want := 30 * time.Second - if cfg.Timeouts.AckWait != want { - t.Fatalf("expected timeouts.ack_wait to default to %v, got %v", want, cfg.Timeouts.AckWait) - } - if cfg.NATS.ConsumerRules.AckWait != want { - t.Fatalf("expected consumer ack_wait to default to %v, got %v", want, cfg.NATS.ConsumerRules.AckWait) - } -} + It("defaults ack waits when not provided", func() { + cfg, err := LoadConfig() + Expect(err).NotTo(HaveOccurred()) + + want := 30 * time.Second + Expect(cfg.Timeouts.AckWait).To(Equal(want)) + Expect(cfg.NATS.ConsumerRules.AckWait).To(Equal(want)) + }) +}) diff --git a/internal/infra/config/suite_test.go b/internal/infra/config/suite_test.go new file mode 100644 index 0000000..0363029 --- /dev/null +++ b/internal/infra/config/suite_test.go @@ -0,0 +1,13 @@ +package config + +import ( + "testing" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +func TestInfraConfig(t *testing.T) { + RegisterFailHandler(Fail) + RunSpecs(t, "Infra Config Suite") +} diff --git a/internal/infra/nats/consumer.go b/internal/infra/nats/consumer.go index 556575e..f6bcd1e 100644 --- a/internal/infra/nats/consumer.go +++ b/internal/infra/nats/consumer.go @@ -6,9 +6,14 @@ import ( "context" "errors" "fmt" + "strings" "time" "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" ) @@ -21,6 +26,11 @@ type Consumer struct { logger *zap.Logger subject string consumerName string + meter metric.Meter + ackPending metric.Int64Histogram + redelivered metric.Int64Histogram + pending metric.Int64Histogram + delivered metric.Int64Histogram } // ProvideConsumer creates a NATS consumer @@ -47,6 +57,7 @@ func ProvideConsumer( subject: subject, consumerName: consumerName, } + consumer.initMetrics() // Create consumer if it doesn't exist if err := consumer.ensureConsumer(); err != nil { @@ -74,12 +85,28 @@ func (c *Consumer) ensureConsumer() error { consumerConfig := &nats.ConsumerConfig{ Durable: c.consumerName, - DeliverPolicy: nats.DeliverAllPolicy, + DeliverPolicy: mapDeliverPolicy(c.cfg.NATS.ConsumerRules.DeliverPolicy), AckPolicy: nats.AckExplicitPolicy, AckWait: ackWait, + ReplayPolicy: mapReplayPolicy(c.cfg.NATS.ConsumerRules.ReplayPolicy), MaxDeliver: c.cfg.NATS.ConsumerRules.MaxDeliver, MaxAckPending: c.cfg.NATS.ConsumerRules.MaxAckPending, FilterSubject: c.subject, + BackOff: c.cfg.NATS.ConsumerRules.Backoff, + } + if consumerConfig.DeliverPolicy == nats.DeliverByStartSequencePolicy && c.cfg.NATS.ConsumerRules.StartSequence > 0 { + consumerConfig.OptStartSeq = c.cfg.NATS.ConsumerRules.StartSequence + } + if consumerConfig.DeliverPolicy == nats.DeliverByStartTimePolicy && strings.TrimSpace(c.cfg.NATS.ConsumerRules.StartTime) != "" { + startTime, err := time.Parse(time.RFC3339, c.cfg.NATS.ConsumerRules.StartTime) + if err != nil { + c.logger.Warn("Invalid start time, falling back to deliver policy defaults", + zap.String("start_time", c.cfg.NATS.ConsumerRules.StartTime), + zap.Error(err), + ) + } else { + consumerConfig.OptStartTime = &startTime + } } _, err := c.js.AddConsumer(streamName, consumerConfig) @@ -93,6 +120,8 @@ func (c *Consumer) ensureConsumer() error { zap.String("stream", streamName), zap.String("subject", c.subject), zap.Duration("ack_wait", ackWait), + zap.String("deliver_policy", c.cfg.NATS.ConsumerRules.DeliverPolicy), + zap.String("replay_policy", c.cfg.NATS.ConsumerRules.ReplayPolicy), ) } @@ -133,6 +162,9 @@ func (c *Consumer) Start(ctx context.Context) error { zap.Duration("batch_timeout", batchTimeout), zap.Int("max_deliver", c.cfg.NATS.ConsumerRules.MaxDeliver), zap.Duration("ack_wait", c.cfg.NATS.ConsumerRules.AckWait), + zap.String("deliver_policy", c.cfg.NATS.ConsumerRules.DeliverPolicy), + zap.String("replay_policy", c.cfg.NATS.ConsumerRules.ReplayPolicy), + zap.Int("backoff_steps", len(c.cfg.NATS.ConsumerRules.Backoff)), ) statsCtx, statsCancel := context.WithCancel(ctx) @@ -177,8 +209,8 @@ func (c *Consumer) Start(ctx context.Context) error { continue } - // Transient error: request redelivery - if nakErr := msg.Nak(); nakErr != nil { + // Transient error: request redelivery with optional delay + if nakErr := c.nakWithStrategy(msg); nakErr != nil { c.logger.Error("Failed to NAK message", zap.Error(nakErr)) } continue @@ -221,6 +253,7 @@ func (c *Consumer) emitConsumerStats(ctx context.Context, streamName string) { zap.Uint64("delivered_consumer_seq", uint64(info.Delivered.Consumer)), zap.Uint64("delivered_stream_seq", uint64(info.Delivered.Stream)), ) + c.recordConsumerMetrics(ctx, info) } } } @@ -254,10 +287,80 @@ func (c *Consumer) Shutdown(ctx context.Context) error { } } +func (c *Consumer) initMetrics() { + meter := otel.Meter("caatsm/nats") + c.meter = meter + + if hist, err := meter.Int64Histogram("nats.consumer.ack_pending"); err == nil { + c.ackPending = hist + } + if hist, err := meter.Int64Histogram("nats.consumer.redelivered"); err == nil { + c.redelivered = hist + } + if hist, err := meter.Int64Histogram("nats.consumer.pending"); err == nil { + c.pending = hist + } + if hist, err := meter.Int64Histogram("nats.consumer.delivered"); err == nil { + c.delivered = hist + } +} + +func (c *Consumer) recordConsumerMetrics(ctx context.Context, info *nats.ConsumerInfo) { + if info == nil { + return + } + if c.ackPending != nil { + c.ackPending.Record(ctx, int64(info.NumAckPending)) + } + if c.redelivered != nil { + c.redelivered.Record(ctx, int64(info.NumRedelivered)) + } + if c.pending != nil { + c.pending.Record(ctx, int64(info.NumPending)) + } + if c.delivered != nil { + c.delivered.Record(ctx, int64(info.Delivered.Stream)) + } +} + +func (c *Consumer) nakWithStrategy(msg *nats.Msg) error { + backoff := c.cfg.NATS.ConsumerRules.Backoff + if len(backoff) == 0 { + return msg.Nak() + } + + meta, err := msg.Metadata() + if err != nil { + c.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(backoff) { + index = len(backoff) - 1 + } + delay := backoff[index] + if delay <= 0 { + return msg.Nak() + } + + return msg.NakWithDelay(delay) +} + // processMessage processes a single message func (c *Consumer) processMessage(ctx context.Context, msg *nats.Msg) error { + ctx, span := otel.Tracer("caatsm/nats").Start(ctx, "Consumer.processMessage") + defer span.End() + span.SetAttributes(attribute.String("nats.subject", msg.Subject)) + msgID, source, err := c.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" { @@ -276,9 +379,12 @@ func (c *Consumer) processMessage(ctx context.Context, msg *nats.Msg) error { // Call processor if err := c.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 } @@ -294,3 +400,29 @@ func (c *Consumer) resolveMsgID(msg *nats.Msg) (string, string, error) { return fmt.Sprintf("js-%d", meta.Sequence.Stream), "metadata", nil } + +func mapDeliverPolicy(value string) nats.DeliverPolicy { + switch strings.ToLower(value) { + case "new": + return nats.DeliverNewPolicy + case "last": + return nats.DeliverLastPolicy + case "last_per_subject": + return nats.DeliverLastPerSubjectPolicy + case "sequence": + return nats.DeliverByStartSequencePolicy + case "time": + return nats.DeliverByStartTimePolicy + default: + return nats.DeliverAllPolicy + } +} + +func mapReplayPolicy(value string) nats.ReplayPolicy { + switch strings.ToLower(value) { + case "original": + return nats.ReplayOriginalPolicy + default: + return nats.ReplayInstantPolicy + } +} diff --git a/internal/infra/postgres/repository.go b/internal/infra/postgres/repository.go index b678a9f..9be97d4 100644 --- a/internal/infra/postgres/repository.go +++ b/internal/infra/postgres/repository.go @@ -5,10 +5,16 @@ import ( "caatsm/internal/adapter/mapper" "caatsm/internal/domain" "context" + "encoding/json" "fmt" + "time" + "github.com/google/uuid" "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgxpool" + "go.opentelemetry.io/otel" + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/codes" "go.uber.org/zap" ) @@ -30,8 +36,14 @@ 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 { + ctx, span := otel.Tracer("caatsm/postgres").Start(ctx, "Repository.InsertOne") + defer span.End() + span.SetAttributes(attribute.String("db.table", "aviation.telegrams")) + row, err := r.mapper.ToDBRow(msg) if err != nil { + span.RecordError(err) + span.SetStatus(codes.Error, err.Error()) return fmt.Errorf("failed to map message to DB row: %w", err) } @@ -39,18 +51,21 @@ func (r *Repository) InsertOne(ctx context.Context, msg *domain.ParsedMessage) e INSERT INTO aviation.telegrams ( uuid, message_id, date_time, priority_indicator, primary_address, secondary_addresses, originator, originator_date_time, category, - content, body_data, received_at, parsed_at, dispatched_at, need_dispatch + content, body_data, status, error_reason, + received_at, parsed_at, dispatched_at, need_dispatch ) VALUES ( - $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15 + $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17 ) ON CONFLICT (uuid) DO NOTHING ` tag, err := r.pool.Exec(ctx, query, row[0], row[1], row[2], row[3], row[4], row[5], row[6], row[7], row[8], - row[9], row[10], row[11], row[12], row[13], row[14], + row[9], row[10], row[11], row[12], row[13], row[14], row[15], row[16], ) if err != nil { + span.RecordError(err) + span.SetStatus(codes.Error, err.Error()) return fmt.Errorf("failed to insert message: %w", err) } @@ -72,6 +87,10 @@ 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 { + ctx, span := otel.Tracer("caatsm/postgres").Start(ctx, "Repository.InsertBatch") + defer span.End() + span.SetAttributes(attribute.String("db.table", "aviation.telegrams")) + if len(msgs) == 0 { return nil } @@ -93,14 +112,18 @@ func (r *Repository) InsertBatch(ctx context.Context, msgs []*domain.ParsedMessa []string{ "uuid", "message_id", "date_time", "priority_indicator", "primary_address", "secondary_addresses", "originator", "originator_date_time", "category", - "content", "body_data", "received_at", "parsed_at", "dispatched_at", "need_dispatch", + "content", "body_data", "status", "error_reason", + "received_at", "parsed_at", "dispatched_at", "need_dispatch", }, pgx.CopyFromRows(rows), ) if err != nil { + span.RecordError(err) + span.SetStatus(codes.Error, err.Error()) return fmt.Errorf("failed to batch insert messages: %w", err) } + span.SetAttributes(attribute.Int64("db.inserted", copyCount)) r.logger.Info("Batch inserted messages", zap.Int("count", int(copyCount)), zap.Int("attempted", len(msgs)), @@ -108,3 +131,69 @@ func (r *Repository) InsertBatch(ctx context.Context, msgs []*domain.ParsedMessa return nil } + +// InsertRaw inserts a failed telegram into aviation.telegrams_raw for post-processing. +func (r *Repository) InsertRaw(ctx context.Context, msg *domain.ParsedMessage) error { + ctx, span := otel.Tracer("caatsm/postgres").Start(ctx, "Repository.InsertRaw") + defer span.End() + span.SetAttributes(attribute.String("db.table", "aviation.telegrams_raw")) + + if msg == nil { + err := fmt.Errorf("message is nil") + span.RecordError(err) + span.SetStatus(codes.Error, err.Error()) + return fmt.Errorf("message is nil") + } + if msg.Uuid == "" { + msg.Uuid = uuid.NewString() + } + if msg.ReceivedAt.IsZero() { + msg.ReceivedAt = time.Now() + } + + metadata := map[string]interface{}{ + "message_id": msg.MessageID, + "category": msg.Category, + "comments": msg.Comments, + } + metadataJSON, err := json.Marshal(metadata) + if err != nil { + return fmt.Errorf("failed to marshal metadata: %w", err) + } + + query := ` + INSERT INTO aviation.telegrams_raw ( + uuid, status, error_reason, content, received_at, metadata + ) VALUES ( + $1, $2, $3, $4, $5, $6 + ) + ON CONFLICT (uuid) DO UPDATE + SET status = EXCLUDED.status, + error_reason = EXCLUDED.error_reason, + content = EXCLUDED.content, + received_at = EXCLUDED.received_at, + metadata = EXCLUDED.metadata + ` + + _, err = r.pool.Exec(ctx, query, + msg.Uuid, + string(msg.Status), + msg.ErrorReason, + msg.Content, + msg.ReceivedAt, + metadataJSON, + ) + if err != nil { + span.RecordError(err) + span.SetStatus(codes.Error, err.Error()) + return fmt.Errorf("failed to insert raw telegram: %w", err) + } + + span.SetAttributes(attribute.String("telegram.uuid", msg.Uuid), attribute.String("telegram.status", string(msg.Status))) + r.logger.Debug("Persisted raw telegram", + zap.String("uuid", msg.Uuid), + zap.String("status", string(msg.Status)), + ) + + return nil +} diff --git a/internal/parsers/aviation_parser.go b/internal/parsers/aviation_parser.go index fb5a80a..facae16 100644 --- a/internal/parsers/aviation_parser.go +++ b/internal/parsers/aviation_parser.go @@ -3,6 +3,7 @@ package parsers import ( "caatsm/internal/domain" "caatsm/pkg/utils" + "errors" "fmt" "regexp" "strings" @@ -51,6 +52,10 @@ var ( performancePattern, regPattern, reroutePattern} + // ErrHeaderParse indicates an invalid header section. + ErrHeaderParse = errors.New("invalid telegram header") + // ErrBodyParse indicates a failure matching the telegram body. + ErrBodyParse = errors.New("invalid telegram body") ) type BodyParser struct { @@ -194,27 +199,33 @@ func (parser *BodyParser) createBodyData(data map[string]string) (string, interf } } -func Parse(rawText string) *domain.ParsedMessage { +func Parse(rawText string) (*domain.ParsedMessage, error) { message, err := ParseHeader(rawText) if err != nil { msg := domain.NewParsedMessage() msg.Content = rawText - return msg + msg.Comments = err.Error() + msg.ErrorReason = err.Error() + msg.Status = domain.MessageStatusHeaderError + return msg, fmt.Errorf("%w: %w", ErrHeaderParse, err) } bodyParser := NewBodyParser(message.Body) - category, bodyData, err := bodyParser.Parse() + category, bodyData, bodyErr := bodyParser.Parse() message.Category = category message.ParsedAt = time.Now() - if err != nil { - message.Comments = err.Error() - return &message + if bodyErr != nil { + message.Comments = bodyErr.Error() + message.ErrorReason = bodyErr.Error() + message.Status = domain.MessageStatusBodyError + return &message, fmt.Errorf("%w: %w", ErrBodyParse, bodyErr) } message.Parsed = true + message.Status = domain.MessageStatusParsed message.BodyData = bodyData message.Uuid = uuid.New().String() - return &message + return &message, nil } func cleanMessage(text string) string { diff --git a/internal/parsers/aviation_parser_test.go b/internal/parsers/aviation_parser_test.go index effb309..38638d8 100644 --- a/internal/parsers/aviation_parser_test.go +++ b/internal/parsers/aviation_parser_test.go @@ -282,7 +282,8 @@ FF ZBTJZPZX NNNN ` It("should parse the whole message correctly", func() { - parsedMessage := Parse(message) + parsedMessage, err := Parse(message) + Expect(err).ToNot(HaveOccurred()) Expect(parsedMessage).ToNot(BeNil()) Expect(parsedMessage.Parsed).To(BeTrue()) Expect(parsedMessage.MessageID).To(Equal("TMQ2526")) @@ -333,7 +334,8 @@ GG ZBTJZPZX NNNN ` It("should parse the whole message correctly", func() { - parsedMessage := Parse(message) + parsedMessage, err := Parse(message) + Expect(err).ToNot(HaveOccurred()) Expect(parsedMessage).ToNot(BeNil()) Expect(parsedMessage.Parsed).To(BeTrue()) Expect(parsedMessage.MessageID).To(Equal("TMQ2617")) diff --git a/internal/repository/telegrams.ddl b/internal/repository/telegrams.ddl index fb4d589..9bbcbef 100644 --- a/internal/repository/telegrams.ddl +++ b/internal/repository/telegrams.ddl @@ -12,6 +12,8 @@ CREATE TABLE aviation.telegrams ( category VARCHAR(255), content TEXT, body_data JSONB, + status VARCHAR(32) NOT NULL DEFAULT 'parsed', + error_reason TEXT, received_at TIMESTAMP NOT NULL, parsed_at TIMESTAMP, dispatched_at TIMESTAMP, @@ -24,3 +26,12 @@ 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 TABLE IF NOT EXISTS aviation.telegrams_raw ( + uuid UUID PRIMARY KEY, + status VARCHAR(32) NOT NULL, + error_reason TEXT, + content TEXT NOT NULL, + received_at TIMESTAMP NOT NULL, + metadata JSONB +); diff --git a/pkg/di/init_manual.go b/pkg/di/init_manual.go deleted file mode 100644 index d64eafa..0000000 --- a/pkg/di/init_manual.go +++ /dev/null @@ -1,60 +0,0 @@ -package di - -import ( - "caatsm/internal/adapter/parser" - "caatsm/internal/app" - "caatsm/internal/infra/config" - "caatsm/internal/infra/log" - "caatsm/internal/infra/nats" - "caatsm/internal/infra/postgres" - "errors" -) - -// InitializeAppWithConfig wires dependencies using the provided config. -func InitializeAppWithConfig(cfg *config.Config) (*app.MessageProcessor, *nats.Consumer, error) { - if cfg == nil { - return nil, nil, errors.New("config is required") - } - - logger, err := log.ProvideLogger(cfg) - if err != nil { - return nil, nil, err - } - - pool, err := postgres.ProvideDB(cfg) - if err != nil { - return nil, nil, err - } - - repo, err := postgres.ProvideRepository(pool, logger) - if err != nil { - return nil, nil, err - } - - conn, err := nats.ProvideNATSConn(cfg, logger) - { - if err != nil { - return nil, nil, err - } - } - - js, err := nats.ProvideJetStream(conn, cfg, logger) - if err != nil { - return nil, nil, err - } - - publisher, err := nats.ProvidePublisher(js, cfg, logger) - if err != nil { - return nil, nil, err - } - - messageParser := parser.ProvideParser() - processor := app.NewMessageProcessor(messageParser, repo, publisher, logger) - - consumer, err := nats.ProvideConsumer(conn, js, processor, cfg, logger) - if err != nil { - return nil, nil, err - } - - return processor, consumer, nil -} diff --git a/pkg/di/wire.go b/pkg/di/wire.go index a2e48da..3db4401 100644 --- a/pkg/di/wire.go +++ b/pkg/di/wire.go @@ -16,30 +16,63 @@ import ( // InitializeApp initializes the application with all dependencies func InitializeApp() (*app.MessageProcessor, *nats.Consumer, error) { - wire.Build( - // Config - config.ProvideConfig, + comps, err := buildAppComponents() + if err != nil { + return nil, nil, err + } + return comps.Processor, comps.Consumer, nil +} + +// InitializeAppWithConfig wires dependencies using a pre-loaded configuration. +func InitializeAppWithConfig(cfg *config.Config) (*app.MessageProcessor, *nats.Consumer, error) { + comps, err := buildAppComponentsWithConfig(cfg) + if err != nil { + return nil, nil, err + } + return comps.Processor, comps.Consumer, nil +} + +var runtimeSet = wire.NewSet( + // Logger + log.ProvideLogger, + + // Database + postgres.ProvideDB, + postgres.ProvideRepository, - // Logger - log.ProvideLogger, + // NATS + nats.ProvideNATSConn, + nats.ProvideJetStream, + nats.ProvidePublisher, - // Database - postgres.ProvideDB, - postgres.ProvideRepository, + // Parser + parser.ProvideParser, - // NATS - nats.ProvideNATSConn, - nats.ProvideJetStream, - nats.ProvidePublisher, + // App + app.NewMessageProcessor, - // Parser - parser.ProvideParser, + // Consumer + nats.ProvideConsumer, +) - // App - app.NewMessageProcessor, +func buildAppComponents() (*appComponents, error) { + wire.Build( + config.ProvideConfig, + runtimeSet, + wire.Struct(new(appComponents), "*"), + ) + return nil, nil +} - // Consumer - nats.ProvideConsumer, +func buildAppComponentsWithConfig(cfg *config.Config) (*appComponents, error) { + wire.Build( + runtimeSet, + wire.Struct(new(appComponents), "*"), ) - return nil, nil, nil + return nil, nil +} + +type appComponents struct { + Processor *app.MessageProcessor + Consumer *nats.Consumer } diff --git a/pkg/di/wire_gen.go b/pkg/di/wire_gen.go index 2c2defe..ba0796b 100644 --- a/pkg/di/wire_gen.go +++ b/pkg/di/wire_gen.go @@ -1,5 +1,6 @@ // Code generated by Wire. DO NOT EDIT. +//go:generate go run -mod=mod github.com/google/wire/cmd/wire //go:build !wireinject // +build !wireinject @@ -12,44 +13,114 @@ import ( "caatsm/internal/infra/log" "caatsm/internal/infra/nats" "caatsm/internal/infra/postgres" + "github.com/google/wire" ) +// Injectors from wire.go: + +func buildAppComponents() (*appComponents, error) { + parserParser := parser.ProvideParser() + configConfig, err := config.ProvideConfig() + if err != nil { + return nil, err + } + pool, err := postgres.ProvideDB(configConfig) + if err != nil { + return nil, err + } + logger, err := log.ProvideLogger(configConfig) + if err != nil { + return nil, err + } + repository, err := postgres.ProvideRepository(pool, logger) + if err != nil { + return nil, err + } + conn, err := nats.ProvideNATSConn(configConfig, logger) + if err != nil { + return nil, err + } + jetStreamContext, err := nats.ProvideJetStream(conn, configConfig, logger) + if err != nil { + return nil, err + } + publisher, err := nats.ProvidePublisher(jetStreamContext, configConfig, logger) + if err != nil { + return nil, err + } + messageProcessor := app.NewMessageProcessor(parserParser, repository, publisher, logger) + consumer, err := nats.ProvideConsumer(conn, jetStreamContext, messageProcessor, configConfig, logger) + if err != nil { + return nil, err + } + diAppComponents := &appComponents{ + Processor: messageProcessor, + Consumer: consumer, + } + return diAppComponents, nil +} + +func buildAppComponentsWithConfig(cfg *config.Config) (*appComponents, error) { + parserParser := parser.ProvideParser() + pool, err := postgres.ProvideDB(cfg) + if err != nil { + return nil, err + } + logger, err := log.ProvideLogger(cfg) + if err != nil { + return nil, err + } + repository, err := postgres.ProvideRepository(pool, logger) + if err != nil { + return nil, err + } + conn, err := nats.ProvideNATSConn(cfg, logger) + if err != nil { + return nil, err + } + jetStreamContext, err := nats.ProvideJetStream(conn, cfg, logger) + if err != nil { + return nil, err + } + publisher, err := nats.ProvidePublisher(jetStreamContext, cfg, logger) + if err != nil { + return nil, err + } + messageProcessor := app.NewMessageProcessor(parserParser, repository, publisher, logger) + consumer, err := nats.ProvideConsumer(conn, jetStreamContext, messageProcessor, cfg, logger) + if err != nil { + return nil, err + } + diAppComponents := &appComponents{ + Processor: messageProcessor, + Consumer: consumer, + } + return diAppComponents, nil +} + +// wire.go: + // InitializeApp initializes the application with all dependencies func InitializeApp() (*app.MessageProcessor, *nats.Consumer, error) { - configConfig, err := config.ProvideConfig() + comps, err := buildAppComponents() if err != nil { return nil, nil, err } - zapLogger, err := log.ProvideLogger(configConfig) - if err != nil { - return nil, nil, err - } - pgxpoolPool, err := postgres.ProvideDB(configConfig) - if err != nil { - return nil, nil, err - } - adapterRepository, err := postgres.ProvideRepository(pgxpoolPool, zapLogger) - if err != nil { - return nil, nil, err - } - natsConn, err := nats.ProvideNATSConn(configConfig, zapLogger) - if err != nil { - return nil, nil, err - } - natsJetStreamContext, err := nats.ProvideJetStream(natsConn, configConfig, zapLogger) - if err != nil { - return nil, nil, err - } - adapterPublisher, err := nats.ProvidePublisher(natsJetStreamContext, configConfig, zapLogger) - if err != nil { - return nil, nil, err - } - parserParser := parser.ProvideParser() - appMessageProcessor := app.NewMessageProcessor(parserParser, adapterRepository, adapterPublisher, zapLogger) - natsConsumer, err := nats.ProvideConsumer(natsConn, natsJetStreamContext, appMessageProcessor, configConfig, zapLogger) - if err != nil { - return nil, nil, err - } - return appMessageProcessor, natsConsumer, nil + return comps.Processor, comps.Consumer, nil } +// InitializeAppWithConfig wires dependencies using a pre-loaded configuration. +func InitializeAppWithConfig(cfg *config.Config) (*app.MessageProcessor, *nats.Consumer, error) { + comps, err := buildAppComponentsWithConfig(cfg) + if err != nil { + return nil, nil, err + } + return comps.Processor, comps.Consumer, nil +} + +var runtimeSet = wire.NewSet(log.ProvideLogger, postgres.ProvideDB, postgres.ProvideRepository, nats.ProvideNATSConn, nats.ProvideJetStream, nats.ProvidePublisher, parser.ProvideParser, app.NewMessageProcessor, nats.ProvideConsumer) + +type appComponents struct { + Processor *app.MessageProcessor + Consumer *nats.Consumer +}