diff --git a/README.md b/README.md new file mode 100644 index 0000000..339556c --- /dev/null +++ b/README.md @@ -0,0 +1,500 @@ +# go-caatsm + +Civil Aviation Authority Telegram Message Processor + +A high-performance, production-ready message processing system for aviation telegrams using Clean Architecture, NATS JetStream, and PostgreSQL. + +## Architecture + +This project follows Clean Architecture principles with clear separation of concerns: + +``` +/cmd/main/main.go # Application entry point +/internal + /app # Application layer (business logic orchestration) + processor.go # Message processor + /domain # Domain models (pure Go types, no external dependencies) + aviation.go # ParsedMessage and related types + /adapter # Adapter layer (interfaces and implementations) + /parser # Message parsing adapters + /mapper # Data mapping (domain ↔ infrastructure) + repository.go # Repository interface + publisher.go # Publisher interface + /infra # Infrastructure layer + /config # Configuration management (Koanf) + /nats # NATS JetStream client + /postgres # PostgreSQL repository (pgx) + /log # Logging (Zap) +/pkg/di # Dependency injection (Wire) +``` + +## Features + +- **Clean Architecture**: Clear separation between domain, application, and infrastructure layers +- **NATS JetStream**: Reliable message streaming with automatic retries and dead-letter queues +- **PostgreSQL**: High-performance data persistence using pgx with batch operations +- **Dependency Injection**: Google Wire for compile-time dependency injection +- **Configuration Management**: Koanf for flexible configuration loading (file + environment variables) +- **Structured Logging**: Zap logger with configurable levels and formats +- **Batch Processing**: Efficient batch message processing and database inserts + +## Prerequisites + +- Go 1.22+ +- PostgreSQL 12+ +- NATS Server with JetStream enabled + +## Installation + +1. Clone the repository: +```bash +git clone +cd go-caatsm +``` + +2. Install dependencies: +```bash +go mod download +``` + +3. Set up PostgreSQL database: +```bash +psql -U postgres -f internal/repository/telegrams.ddl +``` + +4. Configure the application: + - Copy `configs/config.dev.toml` and modify as needed + - Or set environment variables with `CAATSM_` prefix + +## Configuration + +Configuration is loaded from TOML files and environment variables. The configuration file should be located at `configs/config.{env}.toml` where `{env}` is determined by the `GO_ENV` environment variable (defaults to `dev`). + +### Configuration Structure + +```toml +[nats] +url = "nats://localhost:4222" +stream = "TELEGRAM" +consumer = "telegram-consumer" + +[subscription] +topic = "Telegram.Serial" + +[publisher] +topic = "Telegram.Json" + +[postgres] +url = "postgres://user:password@localhost:5432/aviation?sslmode=disable" +max_conns = 10 +min_conns = 2 + +[app] +batch_size = 50 +batch_timeout = "2s" + +[log] +level = "info" +format = "json" +``` + +### Environment Variables + +You can override any configuration value using environment variables with the `CAATSM_` prefix: + +```bash +export CAATSM_NATS_URL="nats://nats-server:4222" +export CAATSM_POSTGRES_URL="postgres://user:pass@db:5432/aviation" +export CAATSM_LOG_LEVEL="debug" +``` + +Environment variable names are converted from `CAATSM_NATS_URL` to `nats.url` in the configuration. + +## Usage + +### Build + +```bash +go build -o bin/receiver ./cmd/main +``` + +Or using Task: + +```bash +task build +``` + +### Run + +```bash +# Development mode +GO_ENV=dev ./bin/receiver listen + +# Production mode +GO_ENV=prod ./bin/receiver listen +``` + +Or using Task: + +```bash +task run-dev +``` + +### Command Line Options + +```bash +./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") +``` + +## Development + +### Project Structure + +- **Domain Layer** (`internal/domain`): Pure business logic and domain models +- **Application Layer** (`internal/app`): Orchestrates business flows +- **Adapter Layer** (`internal/adapter`): Interfaces and adapters between layers +- **Infrastructure Layer** (`internal/infra`): External concerns (NATS, PostgreSQL, config, logging) + +### Adding New Features + +1. **Domain Changes**: Add to `internal/domain` (no external dependencies) +2. **Business Logic**: Add to `internal/app` +3. **External Integrations**: Add to `internal/infra` +4. **Adapters**: Add to `internal/adapter` to bridge between layers + +### Dependency Injection + +Dependencies are managed using Google Wire. To add a new dependency: + +1. Create a provider function in the appropriate package +2. Add it to `pkg/di/wire.go` +3. Run `wire ./pkg/di` to regenerate `wire_gen.go` + +### Testing + +```bash +# Run all tests +go test ./... + +# Run tests with coverage +task coverage +``` + +## Message Flow + +1. **NATS Consumer** receives raw telegram messages from JetStream +2. **MessageProcessor** orchestrates the processing: + - Parses the message using the Parser adapter + - Stores the parsed message in PostgreSQL via Repository + - Publishes the parsed message to the output topic via Publisher +3. **ACK/NAK** is sent based on processing success/failure +4. **Retry Logic** handles transient failures automatically + +## 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. + +### Message Format + +All telegrams follow this general structure: + +``` +ZCZC + + + + +NNNN +``` + +**Header Fields:** +- `ZCZC`: Start indicator +- `MessageID`: Unique message identifier (e.g., "TMQ1324") +- `DateTime`: Message date and time (e.g., "150631") +- `PriorityIndicator`: Message priority (e.g., "FF", "DD") +- `PrimaryAddress`: Primary recipient address (ICAO code) +- `SecondaryAddresses`: Additional recipient addresses +- `Originator`: Message originator (optional) + +### Supported Message Types + +The parser supports the following message categories: + +#### 1. ARR - Arrival Message + +Arrival messages report aircraft arrival information. + +**Format:** +``` +(ARR-[/]--) +``` + +**Parsed Fields:** +- `category`: "ARR" +- `aircraft_id`: Aircraft identification/flight number +- `ssr_mode_and_code`: SSR mode and code (optional) +- `departure_airport`: Departure airport ICAO code +- `departure_time`: Departure time +- `arrival_airport`: Arrival airport ICAO code +- `arrival_time`: Arrival time +- `estimated_elapsed_time`: Estimated flight duration (optional) +- `alternate_airport`: Alternate airport (optional) +- `other_info`: Additional information (optional) + +**Example:** +``` +ZCZC ARR1234 150631 +FF ZBTJZPZX +150630 ZBACZQZX +(ARR-CCA1234-A1234-ZBTJ1500-ZGGG0135) +NNNN +``` + +#### 2. DEP - Departure Message + +Departure messages report aircraft departure information. + +**Format:** +``` +(DEP-[/]--) +``` + +**Parsed Fields:** +- `category`: "DEP" +- `aircraft_id`: Aircraft identification/flight number +- `ssr_mode_and_code`: SSR mode and code (optional) +- `departure_airport`: Departure airport ICAO code +- `departure_time`: Departure time +- `destination`: Destination airport ICAO code +- `estimated_elapsed_time`: Estimated flight duration +- `alternate_airport`: Alternate airport (optional) +- `other_info`: Additional information (optional) + +**Example:** +``` +ZCZC DEP5678 120915 +DD KLAXZPZX +120914 KSFOZQZX +(DEP-ABC5678-A1234-ZBTJ1440-ZGGG) +NNNN +``` + +#### 3. CNL - Cancellation Message + +Cancellation messages indicate flight cancellations. + +**Format:** +``` +(CNL---) +``` + +**Parsed Fields:** +- `category`: "CNL" +- `aircraft_id`: Aircraft identification/flight number +- `departure_airport`: Departure airport ICAO code +- `destination_airport`: Destination airport ICAO code +- `other_info`: Additional information (optional) + +**Example:** +``` +ZCZC CNL9012 150631 +FF ZBTJZPZX +(CNL-CCA9012-ZBTJ-ZGGG) +NNNN +``` + +#### 4. DLA - Delay Message + +Delay messages report flight delays with new departure times. + +**Format:** +``` +(DLA-[/]-[]-[]) +``` + +**Parsed Fields:** +- `category`: "DLA" +- `aircraft_id`: Aircraft identification/flight number +- `ssr_mode_and_code`: SSR mode and code (optional) +- `departure_airport`: Departure airport ICAO code +- `new_departure_time`: New departure time (optional) +- `arrival_airport`: Arrival airport ICAO code +- `arrival_time`: Estimated arrival time (optional) +- `other_info`: Additional information (optional) + +**Example:** +``` +ZCZC DLA3456 150631 +FF ZBTJZPZX +(DLA-CCA3456-A1234-ZBTJ1600-ZGGG0200) +NNNN +``` + +#### 5. FPL - Flight Plan Message + +Flight plan messages contain detailed flight planning information. + +**Format:** +``` +(FPL-- +-/ +- +- +- +-) +``` + +**Parsed Fields:** +- `category`: "FPL" +- `flight_number`: Flight number +- `reference_data`: Reference data (optional) +- `aircraft_id`: Aircraft identification +- `ssr_mode_and_code`: SSR mode and code +- `flight_rules_and_type`: Flight rules and type +- `cruising_speed_and_level`: Cruising speed and flight level +- `departure_airport`: Departure airport ICAO code +- `departure_time`: Departure time +- `route`: Flight route +- `destination_and_total_time`: Destination and estimated total time +- `alternate_airport`: Alternate airport (optional) +- `estimated_arrival_time`: Estimated arrival time +- `pbn`: Performance-based navigation equipment +- `navigation_equipment`: Navigation equipment +- `estimated_elapsed_time`: Estimated elapsed time +- `selcal_code`: SELCAL code +- `register`: Aircraft registration (optional) +- `performance_category`: Performance category +- `reroute_information`: Reroute information (optional) +- `remarks`: Remarks (optional) + +**Example:** +``` +ZCZC FPL7890 150631 +FF ZBTJZPZX +(FPL-JAE7433-IS +-B744/H-SXIRPZJWY/S +-ZBTJ1755 +-K0926S0920 CG A326 VYK W80 HUR B339 GM A575 MANSA/K0919S0980 +-EDDF0948 EDDK +-EET/ZMUB0100 UNKL0236 +REG/B2422 SEL/JLAD +NAV/RNAV1 RNAV5 RNP4 +RMK/AGCS EQUIPPED) +NNNN +``` + +### Parsing Process + +1. **Header Parsing**: The parser extracts header information including message ID, date/time, priority, and addresses +2. **Category Detection**: The parser identifies the message category from the body content +3. **Body Parsing**: Based on the category, the parser applies the appropriate regex pattern to extract structured data +4. **Data Mapping**: Extracted data is mapped to domain model structures (ARR, DEP, CNL, DLA, or FPL) +5. **Storage**: The parsed message is stored in PostgreSQL with: + - Raw content in `content` field + - Parsed structured data in `body_data` field (JSONB) + - Metadata in dedicated columns + +### Parsed Message Structure + +All parsed messages are stored in the `ParsedMessage` domain model: + +```go +type ParsedMessage struct { + Uuid string // Unique identifier + MessageID string // Telegram message ID + DateTime string // Message date/time + PriorityIndicator string // Priority level + PrimaryAddress string // Primary recipient + SecondaryAddresses string // Secondary recipients + Originator string // Message originator + OriginatorDateTime string // Originator date/time + Category string // Message category (ARR, DEP, CNL, DLA, FPL) + Content string // Raw message content + BodyData interface{} // Parsed body data (ARR, DEP, CNL, DLA, or FPL struct) + ReceivedAt time.Time // Reception timestamp + ParsedAt time.Time // Parsing timestamp + DispatchedAt time.Time // Dispatch timestamp + NeedDispatch bool // Dispatch flag + Parsed bool // Parsing success flag + Comments string // Parsing comments/errors +} +``` + +### Error Handling + +- **Invalid Format**: Messages that don't match expected formats are stored with `Parsed = false` and error details in `Comments` +- **Partial Parsing**: Header parsing failures result in storing raw content only +- **Category Mismatch**: Unsupported categories are logged and stored with parsing errors + +## Database Schema + +The application uses the `aviation.telegrams` table. See `internal/repository/telegrams.ddl` for the schema definition. + +Key fields: +- `uuid`: Primary key (UUID) +- `message_id`: Telegram message ID +- `content`: Raw message content +- `body_data`: Parsed body data (JSONB) +- `received_at`, `parsed_at`, `dispatched_at`: Timestamps + +## Logging + +Logging uses Zap with structured logging. Log levels and format can be configured: + +- **Levels**: `debug`, `info`, `warn`, `error` +- **Formats**: `json` (production) or `console` (development) + +Logs include contextual information: +- Message IDs +- Subject names +- Stream names +- Processing attempts + +## Performance + +- **Batch Processing**: Messages are processed in configurable batches (default: 50) +- **Database Inserts**: Uses PostgreSQL `COPY FROM` for efficient batch inserts +- **Connection Pooling**: Configurable PostgreSQL connection pool +- **JetStream**: Reliable message delivery with automatic retries + +## Troubleshooting + +### Connection Issues + +- **NATS**: Check that NATS server is running and JetStream is enabled +- **PostgreSQL**: Verify database connection string and that the schema exists + +### Message Processing Issues + +- Check logs for parsing errors +- Verify message format matches expected telegram format +- Check database constraints and indexes + +### Configuration Issues + +- Ensure `GO_ENV` is set correctly +- Verify configuration file exists at `configs/config.{env}.toml` +- Check environment variable names use `CAATSM_` prefix + +## Migration from Legacy System + +This project was refactored from: +- **Watermill** → **nats.go JetStream** +- **Hasura GraphQL** → **PostgreSQL pgx** +- **Viper** → **Koanf** +- **Manual DI** → **Google Wire** + +The legacy code has been removed. See the project history for migration details. + +## License + +[Add your license here] + +## Contributing + +[Add contributing guidelines here] + diff --git a/Taskfile.yml b/Taskfile.yml index 1474731..1b98ac4 100644 --- a/Taskfile.yml +++ b/Taskfile.yml @@ -104,25 +104,6 @@ tasks: - echo "Linting code..." - golangci-lint run - install-gq: - desc: Install hasura graphql engine introspection tool - cmds: - - echo "Installing gq..." - - pnpm add -g graphqurl - - schema: - desc: Download the GraphQL schema from Hasura server - cmds: - - echo "Downloading GraphQL schema..." - - > - gq {{.hasura_endpoint}} -H 'X-Hasura-Admin-Secret: {{.hasura_secret}}' --introspect > {{.schema_file}} - - generate: - desc: Generate code using genqlient - cmds: - - echo "Generating code using genqlient..." - - go get github.com/Khan/genqlient/generate - - cd internal/repository && go run github.com/Khan/genqlient && cd ../.. help: desc: Show this help message @@ -140,8 +121,5 @@ tasks: - echo " task fmt - Format the code" - echo " task deps - Install dependencies" - echo " task lint - Lint the code" - - echo " task install-gq - Install hasura graphql engine introspection tool" - - echo " task schema - Download the GraphQL schema from Hasura server" - - echo " task generate - Generate code using genqlient" - echo " task upgrade - Upgrade go dependencies" - echo " task help - Show this help message" diff --git a/cmd/main/main.go b/cmd/main/main.go index 0428acf..7fd0b6a 100644 --- a/cmd/main/main.go +++ b/cmd/main/main.go @@ -1,36 +1,28 @@ package main import ( - "caatsm/internal/config" - "caatsm/internal/nats" - "caatsm/internal/repository" - "caatsm/pkg/utils" - "os" - + "caatsm/pkg/di" + "context" "fmt" + "os" + "os/signal" + "syscall" "github.com/urfave/cli/v2" ) -var ( - cfg *config.Config -) - func main() { app := setupApp() if err := app.Run(os.Args); err != nil { fmt.Printf("Error running application: %v\n", err) + os.Exit(1) } } func setupApp() *cli.App { - app := &cli.App{ + return &cli.App{ Name: "telegram message process", - Usage: "A Civial Aviation Authority Telegram Message Processor", - Before: func(c *cli.Context) error { - - return nil - }, + Usage: "A Civil Aviation Authority Telegram Message Processor", Commands: []*cli.Command{ { Name: "listen", @@ -55,38 +47,43 @@ func setupApp() *cli.App { }, }, } - return app -} - -func overrideConfig(c *cli.Context) { - if c.IsSet("nats") { - cfg.Nats.URL = c.String("nats") - fmt.Printf("Overriding nats url to %s\n", cfg.Nats.URL) - } - if c.IsSet("topic") { - cfg.Subscription.Topic = c.String("topic") - fmt.Printf("Overriding nats topic to %s\n", cfg.Subscription.Topic) - } } func executeListen(c *cli.Context) error { - cfg, err := config.LoadConfig() + // Initialize dependencies using Wire + processor, consumer, err := di.InitializeApp() if err != nil { - fmt.Printf("Error loading configuration: %v\n", err) + return fmt.Errorf("failed to initialize app: %w", err) + } + + // Create context with cancellation + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + // Handle graceful shutdown + sigChan := make(chan os.Signal, 1) + signal.Notify(sigChan, os.Interrupt, syscall.SIGTERM) + + // Start consumer in a goroutine + errChan := make(chan error, 1) + go func() { + if err := consumer.Start(ctx); err != nil { + errChan <- fmt.Errorf("consumer error: %w", err) + } + }() + + // Wait for signal or error + select { + case sig := <-sigChan: + fmt.Printf("Received signal: %v, shutting down...\n", sig) + cancel() + case err := <-errChan: return err } - if err := config.ValidateConfig(cfg); err != nil { - fmt.Printf("Invalid configuration: %v\n", err) - return err - } - overrideConfig(c) - fmt.Println("Loaded configuration successfully") - log := utils.GetLogger() - log.Info("Starting nats subscriber") - publisher := nats.NewPub(cfg) - repository := repository.NewHasura(cfg) - handler := nats.NewHandler(cfg, publisher, repository) - subscriber := nats.NewSub(cfg) - subscriber.Subscribe(cfg, handler) + + // Note: processor is initialized but not directly used here + // It's used by the consumer internally + _ = processor + return nil } diff --git a/configs/config.dev.toml b/configs/config.dev.toml index be381ba..d40bedb 100644 --- a/configs/config.dev.toml +++ b/configs/config.dev.toml @@ -2,6 +2,8 @@ url = "nats://localhost:4222" client = "serial-client" cluster = "tele-cluster" +stream = "TELEGRAM" +consumer = "telegram-consumer" [subscription] topic = "Telegram.Serial" @@ -16,6 +18,19 @@ reconnect_wait = "5s" close = "10s" ack_wait = "5s" +[postgres] +url = "postgres://user:password@localhost:5432/aviation?sslmode=disable" +max_conns = 10 +min_conns = 2 + +[app] +batch_size = 50 +batch_timeout = "2s" + +[log] +level = "info" +format = "json" + [hasura] endpoint = "http://localhost:8080/v1/graphql" secret = "aviation-test" \ No newline at end of file diff --git a/go.mod b/go.mod index 78eb95e..e611422 100644 --- a/go.mod +++ b/go.mod @@ -1,39 +1,47 @@ module caatsm -go 1.22.5 +go 1.23.0 require ( - github.com/Khan/genqlient v0.7.0 - github.com/ThreeDotsLabs/watermill v1.3.5 - github.com/ThreeDotsLabs/watermill-nats/v2 v2.0.2 github.com/google/uuid v1.6.0 + github.com/google/wire v0.7.0 + github.com/jackc/pgx/v5 v5.7.6 + github.com/knadh/koanf/parsers/toml v0.1.0 + github.com/knadh/koanf/providers/env v1.1.0 + github.com/knadh/koanf/providers/file v1.2.0 + github.com/knadh/koanf/v2 v2.3.0 github.com/nats-io/nats.go v1.36.0 - github.com/onsi/ginkgo/v2 v2.19.1 - github.com/onsi/gomega v1.34.1 + github.com/onsi/ginkgo/v2 v2.25.1 + github.com/onsi/gomega v1.38.2 github.com/spf13/viper v1.19.0 github.com/urfave/cli/v2 v2.27.4 go.uber.org/zap v1.27.0 - golang.org/x/oauth2 v0.22.0 gopkg.in/natefinch/lumberjack.v2 v2.2.1 ) require ( + github.com/Masterminds/semver/v3 v3.4.0 // indirect github.com/cpuguy83/go-md2man/v2 v2.0.4 // indirect - github.com/fsnotify/fsnotify v1.7.0 // indirect - github.com/go-logr/logr v1.4.2 // indirect + github.com/fsnotify/fsnotify v1.9.0 // indirect + github.com/go-logr/logr v1.4.3 // indirect github.com/go-task/slim-sprig/v3 v3.0.0 // indirect - github.com/google/go-cmp v0.6.0 // indirect - github.com/google/pprof v0.0.0-20240727154555-813a5fbdbec8 // 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/hashicorp/hcl v1.0.0 // 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 github.com/klauspost/compress v1.17.9 // indirect - github.com/lithammer/shortuuid/v3 v3.0.7 // indirect + github.com/knadh/koanf/maps v0.1.2 // indirect github.com/magiconair/properties v1.8.7 // indirect + github.com/mitchellh/copystructure v1.2.0 // indirect github.com/mitchellh/mapstructure v1.5.0 // indirect + github.com/mitchellh/reflectwalk v1.0.2 // indirect github.com/nats-io/nkeys v0.4.7 // indirect github.com/nats-io/nuid v1.0.1 // indirect - github.com/oklog/ulid v1.3.1 // indirect + github.com/pelletier/go-toml v1.9.5 // indirect github.com/pelletier/go-toml/v2 v2.2.2 // indirect - github.com/pkg/errors v0.9.1 // indirect github.com/russross/blackfriday/v2 v2.1.0 // indirect github.com/sagikazarmark/locafero v0.6.0 // indirect github.com/sagikazarmark/slog-shim v0.1.0 // indirect @@ -42,15 +50,17 @@ require ( github.com/spf13/cast v1.7.0 // indirect github.com/spf13/pflag v1.0.5 // indirect github.com/subosito/gotenv v1.6.0 // indirect - github.com/vektah/gqlparser/v2 v2.5.16 // indirect github.com/xrash/smetrics v0.0.0-20240521201337-686a1a2994c1 // indirect + go.uber.org/automaxprocs v1.6.0 // indirect go.uber.org/multierr v1.11.0 // indirect - golang.org/x/crypto v0.26.0 // indirect + go.yaml.in/yaml/v3 v3.0.4 // indirect + golang.org/x/crypto v0.41.0 // indirect golang.org/x/exp v0.0.0-20240808152545-0cdaa3abc0fa // indirect - golang.org/x/net v0.28.0 // indirect - golang.org/x/sys v0.24.0 // indirect - golang.org/x/text v0.17.0 // indirect - golang.org/x/tools v0.24.0 // indirect + golang.org/x/net v0.43.0 // indirect + golang.org/x/sync v0.16.0 // indirect + golang.org/x/sys v0.35.0 // indirect + golang.org/x/text v0.28.0 // indirect + golang.org/x/tools v0.36.0 // indirect gopkg.in/ini.v1 v1.67.0 // indirect gopkg.in/yaml.v3 v3.0.1 // indirect ) diff --git a/internal/adapter/mapper/mapper.go b/internal/adapter/mapper/mapper.go new file mode 100644 index 0000000..898f9f6 --- /dev/null +++ b/internal/adapter/mapper/mapper.go @@ -0,0 +1,13 @@ +package mapper + +import "caatsm/internal/domain" + +// Mapper defines the interface for mapping between domain 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) +} + diff --git a/internal/adapter/mapper/telegram.go b/internal/adapter/mapper/telegram.go new file mode 100644 index 0000000..4b2b32f --- /dev/null +++ b/internal/adapter/mapper/telegram.go @@ -0,0 +1,69 @@ +package mapper + +import ( + "caatsm/internal/domain" + "encoding/json" + "fmt" + "github.com/google/uuid" +) + +// TelegramMapper maps between domain.ParsedMessage and database rows +type TelegramMapper struct{} + +// NewTelegramMapper creates a new telegram mapper +func NewTelegramMapper() *TelegramMapper { + return &TelegramMapper{} +} + +// ToDBRow converts a domain.ParsedMessage to a database row representation +func (m *TelegramMapper) ToDBRow(msg *domain.ParsedMessage) ([]interface{}, error) { + // Parse UUID + var msgUUID uuid.UUID + var err error + if msg.Uuid != "" { + msgUUID, err = uuid.Parse(msg.Uuid) + if err != nil { + return nil, fmt.Errorf("invalid UUID: %w", err) + } + } else { + msgUUID = uuid.New() + } + + // Marshal BodyData to JSONB + var bodyDataJSON []byte + if msg.BodyData != nil { + bodyDataJSON, err = json.Marshal(msg.BodyData) + if err != nil { + return nil, fmt.Errorf("failed to marshal body data: %w", err) + } + } + + // SecondaryAddresses is already a string, so we can use it directly + secondaryAddresses := msg.SecondaryAddresses + + return []interface{}{ + msgUUID, // uuid + msg.MessageID, // message_id + msg.DateTime, // date_time + msg.PriorityIndicator, // priority_indicator + msg.PrimaryAddress, // primary_address + secondaryAddresses, // secondary_addresses (TEXT) + msg.Originator, // originator + msg.OriginatorDateTime, // originator_date_time + msg.Category, // category + msg.Content, // content (TEXT, original message) + bodyDataJSON, // body_data (JSONB) + msg.ReceivedAt, // received_at + msg.ParsedAt, // parsed_at + msg.DispatchedAt, // dispatched_at + msg.NeedDispatch, // need_dispatch + }, nil +} + +// FromDBRow converts a database row to a domain.ParsedMessage +func (m *TelegramMapper) FromDBRow(row []interface{}) (*domain.ParsedMessage, error) { + // This is a placeholder - will be implemented if needed for queries + // For now, we only need ToDBRow for inserts + return nil, fmt.Errorf("FromDBRow not implemented") +} + diff --git a/internal/adapter/parser/aviation_parser.go b/internal/adapter/parser/aviation_parser.go new file mode 100644 index 0000000..74a33b7 --- /dev/null +++ b/internal/adapter/parser/aviation_parser.go @@ -0,0 +1,21 @@ +package parser + +import ( + "caatsm/internal/domain" + "caatsm/internal/parsers" +) + +// AviationParser implements the Parser interface using the existing parsers package +type AviationParser struct{} + +// NewAviationParser creates a new aviation parser +func NewAviationParser() *AviationParser { + return &AviationParser{} +} + +// Parse parses a raw message string and returns a ParsedMessage +func (p *AviationParser) Parse(rawText string) *domain.ParsedMessage { + // 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 new file mode 100644 index 0000000..b7d072c --- /dev/null +++ b/internal/adapter/parser/parser.go @@ -0,0 +1,10 @@ +package parser + +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 +} + diff --git a/internal/adapter/parser/provider.go b/internal/adapter/parser/provider.go new file mode 100644 index 0000000..24f0ad1 --- /dev/null +++ b/internal/adapter/parser/provider.go @@ -0,0 +1,7 @@ +package parser + +// ProvideParser creates a parser instance +func ProvideParser() Parser { + return NewAviationParser() +} + diff --git a/internal/adapter/publisher.go b/internal/adapter/publisher.go new file mode 100644 index 0000000..947151d --- /dev/null +++ b/internal/adapter/publisher.go @@ -0,0 +1,8 @@ +package adapter + +// Publisher defines the interface for publishing parsed messages +type Publisher interface { + // Publish publishes a parsed message + Publish(message interface{}) error +} + diff --git a/internal/adapter/repository.go b/internal/adapter/repository.go new file mode 100644 index 0000000..5ca7069 --- /dev/null +++ b/internal/adapter/repository.go @@ -0,0 +1,16 @@ +package adapter + +import ( + "context" + "caatsm/internal/domain" +) + +// Repository defines the interface for message persistence +type Repository interface { + // InsertOne inserts a single telegram message + InsertOne(ctx context.Context, msg *domain.ParsedMessage) error + + // InsertBatch inserts multiple telegram messages in a batch + InsertBatch(ctx context.Context, msgs []*domain.ParsedMessage) error +} + diff --git a/internal/app/processor.go b/internal/app/processor.go new file mode 100644 index 0000000..af3a25f --- /dev/null +++ b/internal/app/processor.go @@ -0,0 +1,80 @@ +package app + +import ( + "context" + "caatsm/internal/adapter" + "caatsm/internal/adapter/parser" + "fmt" + "go.uber.org/zap" +) + +// MessageProcessor handles message processing +type MessageProcessor struct { + parser parser.Parser + repository adapter.Repository + publisher adapter.Publisher + logger *zap.Logger +} + +// NewMessageProcessor creates a new message processor +func NewMessageProcessor( + parser parser.Parser, + repository adapter.Repository, + publisher adapter.Publisher, + logger *zap.Logger, +) *MessageProcessor { + return &MessageProcessor{ + parser: parser, + repository: repository, + publisher: publisher, + logger: logger, + } +} + +// Handle processes a message +func (p *MessageProcessor) Handle(ctx context.Context, raw []byte, msgID string) error { + if raw == nil || len(raw) == 0 { + return fmt.Errorf("empty message") + } + + // Parse the message + parsed := p.parser.Parse(string(raw)) + if parsed == nil { + return fmt.Errorf("parser returned nil") + } + + // Set the message ID from NATS + parsed.Uuid = msgID + + // Log parsing result + if !parsed.Parsed { + p.logger.Info("Message not parsed", + zap.String("msg_id", msgID), + zap.String("content", parsed.Content), + ) + } else { + p.logger.Info("Message parsed successfully", + zap.String("msg_id", msgID), + zap.String("message_id", parsed.MessageID), + zap.String("category", parsed.Category), + ) + } + + // Insert into database + if err := p.repository.InsertOne(ctx, parsed); err != nil { + return fmt.Errorf("failed to insert message: %w", err) + } + + // Publish parsed message + 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), + ) + // Return error to trigger NAK and retry + return fmt.Errorf("failed to publish message: %w", err) + } + + return nil +} diff --git a/internal/infra/config/config.go b/internal/infra/config/config.go new file mode 100644 index 0000000..cc273ac --- /dev/null +++ b/internal/infra/config/config.go @@ -0,0 +1,169 @@ +package config + +import ( + "fmt" + "os" + "strings" + "time" + + "github.com/knadh/koanf/v2" + "github.com/knadh/koanf/parsers/toml" + "github.com/knadh/koanf/providers/file" + envprovider "github.com/knadh/koanf/providers/env" +) + +// Config holds all application configuration +type Config struct { + NATS NATSConfig `koanf:"nats"` + Postgres PostgresConfig `koanf:"postgres"` + App AppConfig `koanf:"app"` + Log LogConfig `koanf:"log"` + Publisher PublisherConfig `koanf:"publisher"` + // Legacy fields for backward compatibility during migration + Subscription SubscriptionConfig `koanf:"subscription"` + Timeouts TimeoutsConfig `koanf:"timeouts"` +} + +// NATSConfig holds NATS/JetStream configuration +type NATSConfig struct { + URL string `koanf:"url"` + Stream string `koanf:"stream"` + Consumer string `koanf:"consumer"` + // Legacy fields + Client string `koanf:"client"` + Cluster string `koanf:"cluster"` +} + +// PostgresConfig holds PostgreSQL configuration +type PostgresConfig struct { + URL string `koanf:"url"` + MaxConns int32 `koanf:"max_conns"` + MinConns int32 `koanf:"min_conns"` +} + +// AppConfig holds application-level configuration +type AppConfig struct { + BatchSize int `koanf:"batch_size"` + BatchTimeout time.Duration `koanf:"batch_timeout"` +} + +// LogConfig holds logging configuration +type LogConfig struct { + Level string `koanf:"level"` + Format string `koanf:"format"` // json or console +} + +// PublisherConfig holds publisher configuration +type PublisherConfig struct { + Topic string `koanf:"topic"` +} + +// SubscriptionConfig holds subscription configuration (legacy) +type SubscriptionConfig struct { + Topic string `koanf:"topic"` + QueueGroup string `koanf:"queue_group"` +} + +// TimeoutsConfig holds timeout configuration (legacy) +type TimeoutsConfig struct { + Server time.Duration `koanf:"server"` + ReconnectWait time.Duration `koanf:"reconnect_wait"` + Close time.Duration `koanf:"close"` + AckWait time.Duration `koanf:"ack_wait"` +} + +// LoadConfig loads configuration from file and environment variables +func LoadConfig() (*Config, error) { + k := koanf.New(".") + + // Determine environment + env := os.Getenv("GO_ENV") + if env == "" { + env = "dev" + } + + // Load from TOML file + configFile := fmt.Sprintf("configs/config.%s.toml", env) + if err := k.Load(file.Provider(configFile), toml.Parser()); err != nil { + return nil, fmt.Errorf("error loading config file '%s': %w", configFile, err) + } + + // Load from environment variables with CAATSM_ prefix + envProvider := envprovider.Provider("CAATSM_", ".", func(s string) string { + // Convert CAATSM_NATS_URL to nats.url + s = strings.TrimPrefix(s, "CAATSM_") + return strings.ToLower(strings.ReplaceAll(s, "_", ".")) + }) + if err := k.Load(envProvider, nil); err != nil { + // Environment variables are optional, so we don't fail if they're not present + // This allows the config to work with just the file + } + + // Unmarshal into Config struct + var cfg Config + if err := k.Unmarshal("", &cfg); err != nil { + return nil, fmt.Errorf("error unmarshaling config: %w", err) + } + + // Set defaults + if cfg.App.BatchSize == 0 { + cfg.App.BatchSize = 50 + } + if cfg.App.BatchTimeout == 0 { + cfg.App.BatchTimeout = 2 * time.Second + } + if cfg.Postgres.MaxConns == 0 { + cfg.Postgres.MaxConns = 10 + } + if cfg.Postgres.MinConns == 0 { + cfg.Postgres.MinConns = 2 + } + if cfg.Log.Level == "" { + cfg.Log.Level = "info" + } + if cfg.Log.Format == "" { + cfg.Log.Format = "json" + } + if cfg.NATS.Stream == "" { + cfg.NATS.Stream = "TELEGRAM" + } + if cfg.NATS.Consumer == "" { + cfg.NATS.Consumer = "telegram-consumer" + } + + // Validate configuration + if err := cfg.Validate(); err != nil { + return nil, fmt.Errorf("config validation failed: %w", err) + } + + return &cfg, nil +} + +// Validate validates the configuration +func (c *Config) Validate() error { + if c.NATS.URL == "" { + return fmt.Errorf("nats.url is required") + } + if c.Subscription.Topic == "" && c.NATS.Stream == "" { + return fmt.Errorf("subscription.topic or nats.stream is required") + } + if c.Publisher.Topic == "" { + return fmt.Errorf("publisher.topic is required") + } + if c.Postgres.URL == "" { + return fmt.Errorf("postgres.url is required") + } + if c.App.BatchSize <= 0 { + return fmt.Errorf("app.batch_size must be greater than 0") + } + if c.App.BatchTimeout <= 0 { + return fmt.Errorf("app.batch_timeout must be greater than 0") + } + return nil +} + +// ProvideConfig is a Wire provider function +func ProvideConfig() (*Config, error) { + return LoadConfig() +} + diff --git a/internal/infra/log/logger.go b/internal/infra/log/logger.go new file mode 100644 index 0000000..2d8c8c5 --- /dev/null +++ b/internal/infra/log/logger.go @@ -0,0 +1,45 @@ +package log + +import ( + "caatsm/internal/infra/config" + "go.uber.org/zap" + "go.uber.org/zap/zapcore" +) + +// ProvideLogger creates a zap logger based on configuration +func ProvideLogger(cfg *config.Config) (*zap.Logger, error) { + var zapConfig zap.Config + + if cfg.Log.Format == "console" { + zapConfig = zap.NewDevelopmentConfig() + } else { + zapConfig = zap.NewProductionConfig() + } + + // Set log level + switch cfg.Log.Level { + case "debug": + zapConfig.Level = zap.NewAtomicLevelAt(zapcore.DebugLevel) + case "info": + zapConfig.Level = zap.NewAtomicLevelAt(zapcore.InfoLevel) + case "warn": + zapConfig.Level = zap.NewAtomicLevelAt(zapcore.WarnLevel) + case "error": + zapConfig.Level = zap.NewAtomicLevelAt(zapcore.ErrorLevel) + default: + zapConfig.Level = zap.NewAtomicLevelAt(zapcore.InfoLevel) + } + + // Build logger + logger, err := zapConfig.Build(zap.AddCaller(), zap.AddStacktrace(zapcore.ErrorLevel)) + if err != nil { + return nil, err + } + + // Note: We still replace global logger for backward compatibility with parsers package + // This will be removed once parsers are fully migrated to use dependency injection + zap.ReplaceGlobals(logger) + + return logger, nil +} + diff --git a/internal/infra/nats/consumer.go b/internal/infra/nats/consumer.go new file mode 100644 index 0000000..5e883ad --- /dev/null +++ b/internal/infra/nats/consumer.go @@ -0,0 +1,184 @@ +package nats + +import ( + "caatsm/internal/app" + "caatsm/internal/infra/config" + "context" + "errors" + "fmt" + "go.uber.org/zap" + "github.com/nats-io/nats.go" + "time" +) + +// Consumer handles NATS JetStream message consumption +type Consumer struct { + js nats.JetStreamContext + processor *app.MessageProcessor + cfg *config.Config + logger *zap.Logger + subject string + consumerName string +} + +// ProvideConsumer creates a NATS consumer +func ProvideConsumer( + js nats.JetStreamContext, + processor *app.MessageProcessor, + cfg *config.Config, + logger *zap.Logger, +) (*Consumer, error) { + subject := cfg.Subscription.Topic + if subject == "" { + subject = "telegram.>" + } + + consumerName := cfg.NATS.Consumer + if consumerName == "" { + consumerName = "telegram-consumer" + } + + consumer := &Consumer{ + js: js, + processor: processor, + cfg: cfg, + logger: logger, + subject: subject, + consumerName: consumerName, + } + + // Create consumer if it doesn't exist + if err := consumer.ensureConsumer(); err != nil { + return nil, fmt.Errorf("failed to ensure consumer: %w", err) + } + + return consumer, nil +} + +// ensureConsumer creates the consumer if it doesn't exist +func (c *Consumer) ensureConsumer() error { + streamName := c.cfg.NATS.Stream + if streamName == "" { + streamName = "TELEGRAM" + } + + consumerConfig := &nats.ConsumerConfig{ + Durable: c.consumerName, + DeliverPolicy: nats.DeliverAllPolicy, + AckPolicy: nats.AckExplicitPolicy, + AckWait: c.cfg.Timeouts.AckWait, + MaxDeliver: 5, // Maximum number of delivery attempts + FilterSubject: c.subject, + } + + _, err := c.js.AddConsumer(streamName, consumerConfig) + if err != nil && err != nats.ErrConsumerNameAlreadyInUse { + return fmt.Errorf("failed to create consumer: %w", err) + } + + if err == nil { + c.logger.Info("Created JetStream consumer", + zap.String("consumer", c.consumerName), + zap.String("stream", streamName), + zap.String("subject", c.subject), + ) + } + + return nil +} + +// Start starts consuming messages +func (c *Consumer) Start(ctx context.Context) error { + streamName := c.cfg.NATS.Stream + if streamName == "" { + streamName = "TELEGRAM" + } + + // Create pull subscription + sub, err := c.js.PullSubscribe(c.subject, c.consumerName, nats.Bind(streamName, c.consumerName)) + if err != nil { + return fmt.Errorf("failed to create pull subscription: %w", err) + } + defer sub.Unsubscribe() + + c.logger.Info("Started consuming messages", + zap.String("subject", c.subject), + zap.String("consumer", c.consumerName), + zap.String("stream", streamName), + ) + + batchSize := c.cfg.App.BatchSize + if batchSize == 0 { + batchSize = 50 + } + batchTimeout := c.cfg.App.BatchTimeout + if batchTimeout == 0 { + batchTimeout = 2 * time.Second + } + + for { + select { + case <-ctx.Done(): + c.logger.Info("Stopping consumer", zap.Error(ctx.Err())) + return ctx.Err() + default: + } + + // Fetch messages in batch + msgs, err := sub.Fetch(batchSize, nats.MaxWait(batchTimeout)) + if err != nil { + if errors.Is(err, nats.ErrTimeout) { + // Timeout is expected when no messages are available + continue + } + c.logger.Error("Failed to fetch messages", zap.Error(err)) + time.Sleep(time.Second) + continue + } + + // Process each message + for _, msg := range msgs { + if err := c.processMessage(ctx, msg); err != nil { + c.logger.Error("Failed to process message", + zap.String("subject", msg.Subject), + zap.Error(err), + ) + // NAK the message to retry + if nakErr := msg.Nak(); nakErr != nil { + c.logger.Error("Failed to NAK message", zap.Error(nakErr)) + } + } else { + // ACK the message + if ackErr := msg.Ack(); ackErr != nil { + c.logger.Error("Failed to ACK message", zap.Error(ackErr)) + } + } + } + } +} + +// processMessage processes a single message +func (c *Consumer) processMessage(ctx context.Context, msg *nats.Msg) error { + msgID := msg.Header.Get("Nats-Msg-Id") + if msgID == "" { + // Use reply subject or generate a simple ID + if msg.Reply != "" { + msgID = msg.Reply + } else { + msgID = fmt.Sprintf("msg-%d", time.Now().UnixNano()) + } + } + + c.logger.Debug("Processing message", + zap.String("subject", msg.Subject), + zap.String("msg_id", msgID), + zap.Int("data_size", len(msg.Data)), + ) + + // Call processor + if err := c.processor.Handle(ctx, msg.Data, msgID); err != nil { + return fmt.Errorf("processor error: %w", err) + } + + return nil +} diff --git a/internal/infra/nats/jetstream.go b/internal/infra/nats/jetstream.go new file mode 100644 index 0000000..b8d0970 --- /dev/null +++ b/internal/infra/nats/jetstream.go @@ -0,0 +1,66 @@ +package nats + +import ( + "caatsm/internal/infra/config" + "fmt" + "go.uber.org/zap" + "github.com/nats-io/nats.go" + "time" +) + +// ProvideJetStream creates a NATS JetStream connection +func ProvideJetStream(cfg *config.Config, logger *zap.Logger) (nats.JetStreamContext, error) { + // Connect to NATS + nc, err := nats.Connect( + cfg.NATS.URL, + nats.RetryOnFailedConnect(true), + nats.Timeout(cfg.Timeouts.Server), + nats.ReconnectWait(cfg.Timeouts.ReconnectWait), + nats.DisconnectErrHandler(func(nc *nats.Conn, err error) { + if err != nil { + logger.Warn("NATS disconnected", zap.Error(err)) + } + }), + nats.ReconnectHandler(func(nc *nats.Conn) { + logger.Info("NATS reconnected", zap.String("url", nc.ConnectedUrl())) + }), + ) + if err != nil { + return nil, fmt.Errorf("failed to connect to NATS: %w", err) + } + + // Get JetStream context + js, err := nc.JetStream() + if err != nil { + nc.Close() + return nil, fmt.Errorf("failed to get JetStream context: %w", err) + } + + // Create stream if it doesn't exist + streamName := cfg.NATS.Stream + subject := cfg.Subscription.Topic + if subject == "" { + subject = "telegram.>" + } + + streamConfig := &nats.StreamConfig{ + Name: streamName, + Subjects: []string{subject}, + Retention: nats.LimitsPolicy, + MaxAge: 24 * time.Hour, + Storage: nats.FileStorage, + Replicas: 1, + } + + _, err = js.AddStream(streamConfig) + if err != nil && err != nats.ErrStreamNameAlreadyInUse { + nc.Close() + return nil, fmt.Errorf("failed to create stream: %w", err) + } + + if err == nil { + logger.Info("Created JetStream", zap.String("stream", streamName), zap.String("subject", subject)) + } + + return js, nil +} diff --git a/internal/infra/nats/publisher.go b/internal/infra/nats/publisher.go new file mode 100644 index 0000000..6e8b63d --- /dev/null +++ b/internal/infra/nats/publisher.go @@ -0,0 +1,57 @@ +package nats + +import ( + "caatsm/internal/adapter" + "caatsm/internal/infra/config" + "encoding/json" + "fmt" + "go.uber.org/zap" + "github.com/nats-io/nats.go" +) + +// Publisher publishes messages to NATS JetStream +type Publisher struct { + js nats.JetStreamContext + cfg *config.Config + logger *zap.Logger +} + +// ProvidePublisher creates a NATS publisher +func ProvidePublisher( + js nats.JetStreamContext, + cfg *config.Config, + logger *zap.Logger, +) (adapter.Publisher, error) { + return &Publisher{ + js: js, + cfg: cfg, + logger: logger, + }, nil +} + +// Publish publishes a message +func (p *Publisher) Publish(message interface{}) error { + topic := p.cfg.Publisher.Topic + if topic == "" { + return fmt.Errorf("publisher topic is not configured") + } + + // Marshal message to JSON + messageBytes, err := json.Marshal(message) + if err != nil { + return fmt.Errorf("failed to marshal message: %w", err) + } + + // Publish to JetStream + _, err = p.js.Publish(topic, messageBytes) + if err != nil { + return fmt.Errorf("failed to publish message: %w", err) + } + + p.logger.Debug("Published message", + zap.String("topic", topic), + zap.Int("size", len(messageBytes)), + ) + + return nil +} diff --git a/internal/infra/postgres/db.go b/internal/infra/postgres/db.go new file mode 100644 index 0000000..38d4379 --- /dev/null +++ b/internal/infra/postgres/db.go @@ -0,0 +1,38 @@ +package postgres + +import ( + "caatsm/internal/infra/config" + "context" + "fmt" + "time" + + "github.com/jackc/pgx/v5/pgxpool" +) + +// ProvideDB creates a PostgreSQL connection pool +func ProvideDB(cfg *config.Config) (*pgxpool.Pool, error) { + ctx := context.Background() + + poolConfig, err := pgxpool.ParseConfig(cfg.Postgres.URL) + if err != nil { + return nil, fmt.Errorf("failed to parse postgres URL: %w", err) + } + + poolConfig.MaxConns = int32(cfg.Postgres.MaxConns) + poolConfig.MinConns = int32(cfg.Postgres.MinConns) + poolConfig.MaxConnLifetime = time.Hour + poolConfig.MaxConnIdleTime = time.Minute * 30 + + pool, err := pgxpool.NewWithConfig(ctx, poolConfig) + if err != nil { + return nil, fmt.Errorf("failed to create connection pool: %w", err) + } + + // Test connection + if err := pool.Ping(ctx); err != nil { + return nil, fmt.Errorf("failed to ping database: %w", err) + } + + return pool, nil +} + diff --git a/internal/infra/postgres/repository.go b/internal/infra/postgres/repository.go new file mode 100644 index 0000000..a56ae12 --- /dev/null +++ b/internal/infra/postgres/repository.go @@ -0,0 +1,101 @@ +package postgres + +import ( + "caatsm/internal/adapter" + "caatsm/internal/adapter/mapper" + "caatsm/internal/domain" + "context" + "fmt" + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" + "go.uber.org/zap" +) + +// Repository implements the adapter.Repository interface using PostgreSQL +type Repository struct { + pool *pgxpool.Pool + mapper *mapper.TelegramMapper + logger *zap.Logger +} + +// ProvideRepository creates a PostgreSQL repository +func ProvideRepository(pool *pgxpool.Pool, logger *zap.Logger) (adapter.Repository, error) { + return &Repository{ + pool: pool, + mapper: mapper.NewTelegramMapper(), + logger: logger, + }, nil +} + +// InsertOne inserts a single telegram message +func (r *Repository) InsertOne(ctx context.Context, msg *domain.ParsedMessage) error { + row, err := r.mapper.ToDBRow(msg) + if err != nil { + return fmt.Errorf("failed to map message to DB row: %w", err) + } + + query := ` + 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 + ) VALUES ( + $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15 + ) + ON CONFLICT (uuid) DO NOTHING + ` + + _, 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], + ) + if err != nil { + return fmt.Errorf("failed to insert message: %w", err) + } + + r.logger.Debug("Inserted message", + zap.String("uuid", msg.Uuid), + zap.String("message_id", msg.MessageID), + ) + + return nil +} + +// InsertBatch inserts multiple telegram messages in a batch using CopyFrom +func (r *Repository) InsertBatch(ctx context.Context, msgs []*domain.ParsedMessage) error { + if len(msgs) == 0 { + return nil + } + + // Convert messages to rows + rows := make([][]interface{}, len(msgs)) + for i, msg := range msgs { + row, err := r.mapper.ToDBRow(msg) + if err != nil { + return fmt.Errorf("failed to map message %d to DB row: %w", i, err) + } + rows[i] = row + } + + // Use CopyFrom for efficient batch insert + copyCount, err := r.pool.CopyFrom( + ctx, + pgx.Identifier{"aviation", "telegrams"}, + []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", + }, + pgx.CopyFromRows(rows), + ) + if err != nil { + return fmt.Errorf("failed to batch insert messages: %w", err) + } + + r.logger.Info("Batch inserted messages", + zap.Int("count", int(copyCount)), + zap.Int("attempted", len(msgs)), + ) + + return nil +} diff --git a/internal/nats/handler.go b/internal/nats/handler.go deleted file mode 100644 index 781c6a9..0000000 --- a/internal/nats/handler.go +++ /dev/null @@ -1,49 +0,0 @@ -package nats - -import ( - "caatsm/internal/config" - "caatsm/internal/domain" - "caatsm/internal/iface" - "caatsm/internal/parsers" - "caatsm/pkg/utils" - "fmt" - "sync" -) - -type MessageHandler struct { - mu sync.Mutex - config *config.Config - repository iface.MessageRepository - publisher iface.MessagePublisher -} - -func NewHandler(config *config.Config, publisher iface.MessagePublisher, repository iface.MessageRepository) *MessageHandler { - return &MessageHandler{ - config: config, - repository: repository, - publisher: publisher, - } -} - -func (handler *MessageHandler) HandleMessage(msg []byte, id string) error { - handler.mu.Lock() - defer handler.mu.Unlock() - log := utils.GetSugaredLogger() - if msg == nil { - log.Error("empty message") - return fmt.Errorf("empty message") - } - payload := string(msg) - var parsed *domain.ParsedMessage - if parsed = parsers.Parse(payload); !parsed.Parsed { - log.Infof("not parsed: [%s] : {%s} \n", id, payload) - } else { - parsed.Uuid = id - log.Infof("parsed [%s]: %v\n", id, parsed.ToString()) - } - handler.repository.CreateNew(parsed) - - handler.publisher.Publish(parsed) - - return nil -} diff --git a/internal/nats/pub.go b/internal/nats/pub.go deleted file mode 100644 index 574042c..0000000 --- a/internal/nats/pub.go +++ /dev/null @@ -1,55 +0,0 @@ -package nats - -import ( - "caatsm/internal/config" - "caatsm/pkg/utils" - "encoding/json" - - "github.com/ThreeDotsLabs/watermill" - "github.com/ThreeDotsLabs/watermill-nats/v2/pkg/nats" - "github.com/ThreeDotsLabs/watermill/message" - nc "github.com/nats-io/nats.go" -) - -type NatsPublisher struct { - config *config.Config - publisher *nats.Publisher -} - -func NewPub(config *config.Config) *NatsPublisher { - logger := watermill.NewStdLogger(false, false) - - jsConfig := nats.JetStreamConfig{Disabled: true} - options := []nc.Option{ - nc.RetryOnFailedConnect(true), - nc.Timeout(config.Timeouts.Server), - nc.ReconnectWait(config.Timeouts.ReconnectWait), - } - publisher, _ := nats.NewPublisher( - nats.PublisherConfig{ - URL: config.Nats.URL, - NatsOptions: options, - JetStream: jsConfig, - }, logger) - return &NatsPublisher{ - config: config, - publisher: publisher, - } -} - -func (n *NatsPublisher) Publish(parsedMessage interface{}) error { - logger := utils.GetSugaredLogger() - - messageText, err := json.Marshal(parsedMessage) - if err != nil { - logger.Errorf("Failed to marshal message: %v", err) - } - msg := message.NewMessage(watermill.NewUUID(), []byte(messageText)) - err = n.publisher.Publish(n.config.Publisher.Topic, msg) - if err != nil { - logger.Errorf("Failed to publish message: %v", err) - return err - } - logger.Infof("Message published: %s", msg.UUID) - return nil -} diff --git a/internal/nats/sub.go b/internal/nats/sub.go deleted file mode 100644 index 56498c2..0000000 --- a/internal/nats/sub.go +++ /dev/null @@ -1,80 +0,0 @@ -package nats - -import ( - "caatsm/internal/config" - "caatsm/internal/iface" - "caatsm/pkg/utils" - "context" - "errors" - - "github.com/ThreeDotsLabs/watermill" - "github.com/ThreeDotsLabs/watermill-nats/v2/pkg/nats" - "github.com/ThreeDotsLabs/watermill/message" - nc "github.com/nats-io/nats.go" -) - -type NatsSubscriber struct { - config *config.Config - subscriber *nats.Subscriber -} - -func NewSub(config *config.Config) *NatsSubscriber { - logger := watermill.NewStdLogger(false, false) - marshaler := &PlainTextMarshaler{} - options := []nc.Option{ - nc.RetryOnFailedConnect(true), - nc.Timeout(config.Timeouts.Server), - nc.ReconnectWait(config.Timeouts.ReconnectWait), - } - jsConfig := nats.JetStreamConfig{Disabled: true} - subscriber, _ := nats.NewSubscriber( - nats.SubscriberConfig{ - URL: config.Nats.URL, - CloseTimeout: config.Timeouts.Close, - AckWaitTimeout: config.Timeouts.AckWait, - NatsOptions: options, - Unmarshaler: marshaler, - JetStream: jsConfig, - }, - logger, - ) - return &NatsSubscriber{ - config: config, - subscriber: subscriber, - } -} - -func (n *NatsSubscriber) Subscribe(config *config.Config, handlers iface.MessageHandler) { - logger := utils.GetSugaredLogger() - - defer n.subscriber.Close() - messages, err := n.subscriber.Subscribe(context.Background(), config.Subscription.Topic) - if err != nil { - logger.Errorf("Failed to subscribe to topic: %v", err) - return - } - - for msg := range messages { - if err := handlers.HandleMessage(msg.Payload, msg.UUID); err == nil { - logger.Infof("Message handled: %s", msg.UUID) - msg.Ack() - } else { - logger.Errorf("Failed to handle message [%s]: %v", msg.UUID, err) - msg.Nack() - } - } -} - -type PlainTextMarshaler struct{} - -func (m *PlainTextMarshaler) Marshal(topic string, msg nc.Msg) ([]byte, error) { - return msg.Data, nil -} - -func (m *PlainTextMarshaler) Unmarshal(newMsg *nc.Msg) (*message.Message, error) { - if newMsg == nil { - return nil, errors.New("empty message") - } - msg := message.NewMessage(watermill.NewUUID(), newMsg.Data) - return msg, nil -} diff --git a/internal/repository/generated.go b/internal/repository/generated.go deleted file mode 100644 index 3667a73..0000000 --- a/internal/repository/generated.go +++ /dev/null @@ -1,154 +0,0 @@ -// Code generated by github.com/Khan/genqlient, DO NOT EDIT. - -package repository - -import ( - "context" - "encoding/json" - "time" - - "github.com/Khan/genqlient/graphql" - "github.com/google/uuid" -) - -// input type for inserting data into table "aviation.telegrams" -type Aviation_telegrams_insert_input struct { - Body_data json.RawMessage `json:"body_data"` - Category string `json:"category"` - Content string `json:"content"` - Date_time string `json:"date_time"` - Dispatched_at time.Time `json:"dispatched_at"` - Message_id string `json:"message_id"` - Need_dispatch bool `json:"need_dispatch"` - Originator string `json:"originator"` - Originator_date_time string `json:"originator_date_time"` - Parsed_at time.Time `json:"parsed_at"` - Primary_address string `json:"primary_address"` - Priority_indicator string `json:"priority_indicator"` - Received_at time.Time `json:"received_at"` - Secondary_addresses string `json:"secondary_addresses"` - Uuid uuid.UUID `json:"uuid"` -} - -// GetBody_data returns Aviation_telegrams_insert_input.Body_data, and is useful for accessing the field via an interface. -func (v *Aviation_telegrams_insert_input) GetBody_data() json.RawMessage { return v.Body_data } - -// GetCategory returns Aviation_telegrams_insert_input.Category, and is useful for accessing the field via an interface. -func (v *Aviation_telegrams_insert_input) GetCategory() string { return v.Category } - -// GetContent returns Aviation_telegrams_insert_input.Content, and is useful for accessing the field via an interface. -func (v *Aviation_telegrams_insert_input) GetContent() string { return v.Content } - -// GetDate_time returns Aviation_telegrams_insert_input.Date_time, and is useful for accessing the field via an interface. -func (v *Aviation_telegrams_insert_input) GetDate_time() string { return v.Date_time } - -// GetDispatched_at returns Aviation_telegrams_insert_input.Dispatched_at, and is useful for accessing the field via an interface. -func (v *Aviation_telegrams_insert_input) GetDispatched_at() time.Time { return v.Dispatched_at } - -// GetMessage_id returns Aviation_telegrams_insert_input.Message_id, and is useful for accessing the field via an interface. -func (v *Aviation_telegrams_insert_input) GetMessage_id() string { return v.Message_id } - -// GetNeed_dispatch returns Aviation_telegrams_insert_input.Need_dispatch, and is useful for accessing the field via an interface. -func (v *Aviation_telegrams_insert_input) GetNeed_dispatch() bool { return v.Need_dispatch } - -// GetOriginator returns Aviation_telegrams_insert_input.Originator, and is useful for accessing the field via an interface. -func (v *Aviation_telegrams_insert_input) GetOriginator() string { return v.Originator } - -// GetOriginator_date_time returns Aviation_telegrams_insert_input.Originator_date_time, and is useful for accessing the field via an interface. -func (v *Aviation_telegrams_insert_input) GetOriginator_date_time() string { - return v.Originator_date_time -} - -// GetParsed_at returns Aviation_telegrams_insert_input.Parsed_at, and is useful for accessing the field via an interface. -func (v *Aviation_telegrams_insert_input) GetParsed_at() time.Time { return v.Parsed_at } - -// GetPrimary_address returns Aviation_telegrams_insert_input.Primary_address, and is useful for accessing the field via an interface. -func (v *Aviation_telegrams_insert_input) GetPrimary_address() string { return v.Primary_address } - -// GetPriority_indicator returns Aviation_telegrams_insert_input.Priority_indicator, and is useful for accessing the field via an interface. -func (v *Aviation_telegrams_insert_input) GetPriority_indicator() string { return v.Priority_indicator } - -// GetReceived_at returns Aviation_telegrams_insert_input.Received_at, and is useful for accessing the field via an interface. -func (v *Aviation_telegrams_insert_input) GetReceived_at() time.Time { return v.Received_at } - -// GetSecondary_addresses returns Aviation_telegrams_insert_input.Secondary_addresses, and is useful for accessing the field via an interface. -func (v *Aviation_telegrams_insert_input) GetSecondary_addresses() string { - return v.Secondary_addresses -} - -// GetUuid returns Aviation_telegrams_insert_input.Uuid, and is useful for accessing the field via an interface. -func (v *Aviation_telegrams_insert_input) GetUuid() uuid.UUID { return v.Uuid } - -// __newMessageInput is used internally by genqlient -type __newMessageInput struct { - Object Aviation_telegrams_insert_input `json:"object"` -} - -// GetObject returns __newMessageInput.Object, and is useful for accessing the field via an interface. -func (v *__newMessageInput) GetObject() Aviation_telegrams_insert_input { return v.Object } - -// newMessageInsert_aviation_telegrams_oneAviation_telegrams includes the requested fields of the GraphQL type aviation_telegrams. -// The GraphQL type's documentation follows. -// -// columns and relationships of "aviation.telegrams" -type newMessageInsert_aviation_telegrams_oneAviation_telegrams struct { - Message_id string `json:"message_id"` - Uuid uuid.UUID `json:"uuid"` -} - -// GetMessage_id returns newMessageInsert_aviation_telegrams_oneAviation_telegrams.Message_id, and is useful for accessing the field via an interface. -func (v *newMessageInsert_aviation_telegrams_oneAviation_telegrams) GetMessage_id() string { - return v.Message_id -} - -// GetUuid returns newMessageInsert_aviation_telegrams_oneAviation_telegrams.Uuid, and is useful for accessing the field via an interface. -func (v *newMessageInsert_aviation_telegrams_oneAviation_telegrams) GetUuid() uuid.UUID { - return v.Uuid -} - -// newMessageResponse is returned by newMessage on success. -type newMessageResponse struct { - // insert a single row into the table: "aviation.telegrams" - Insert_aviation_telegrams_one newMessageInsert_aviation_telegrams_oneAviation_telegrams `json:"insert_aviation_telegrams_one"` -} - -// GetInsert_aviation_telegrams_one returns newMessageResponse.Insert_aviation_telegrams_one, and is useful for accessing the field via an interface. -func (v *newMessageResponse) GetInsert_aviation_telegrams_one() newMessageInsert_aviation_telegrams_oneAviation_telegrams { - return v.Insert_aviation_telegrams_one -} - -// The query or mutation executed by newMessage. -const newMessage_Operation = ` -mutation newMessage ($object: aviation_telegrams_insert_input!) { - insert_aviation_telegrams_one(object: $object) { - message_id - uuid - } -} -` - -func newMessage( - ctx_ context.Context, - client_ graphql.Client, - object Aviation_telegrams_insert_input, -) (*newMessageResponse, error) { - req_ := &graphql.Request{ - OpName: "newMessage", - Query: newMessage_Operation, - Variables: &__newMessageInput{ - Object: object, - }, - } - var err_ error - - var data_ newMessageResponse - resp_ := &graphql.Response{Data: &data_} - - err_ = client_.MakeRequest( - ctx_, - req_, - resp_, - ) - - return &data_, err_ -} diff --git a/internal/repository/genqlient.yaml b/internal/repository/genqlient.yaml deleted file mode 100644 index 265ed81..0000000 --- a/internal/repository/genqlient.yaml +++ /dev/null @@ -1,13 +0,0 @@ -# Default genqlient config; for full documentation see: -# https://github.com/Khan/genqlient/blob/main/docs/genqlient.yaml -schema: schema.graphql -operations: -- genqlient.graphql -generated: generated.go -bindings: - jsonb: - type: encoding/json.RawMessage - timestamp: - type: time.Time - uuid: - type: github.com/google/uuid.UUID \ No newline at end of file diff --git a/internal/repository/hasura.go b/internal/repository/hasura.go deleted file mode 100644 index 8ae82f6..0000000 --- a/internal/repository/hasura.go +++ /dev/null @@ -1,64 +0,0 @@ -package repository - -import ( - "context" - "encoding/json" - "os" - - "caatsm/internal/config" - "caatsm/internal/domain" - "caatsm/pkg/utils" - - "github.com/Khan/genqlient/graphql" - "golang.org/x/oauth2" -) - -type HasuraRepository struct { - client graphql.Client -} - -// New creates a new HasuraRepository -func NewHasura(config *config.Config) *HasuraRepository { - token := os.Getenv("GRAPHQL_TOKEN") - if token == "" { - token = config.Hasura.Secret - } - src := oauth2.StaticTokenSource( - &oauth2.Token{AccessToken: token}, - ) - httpClient := oauth2.NewClient(context.Background(), src) - return &HasuraRepository{ - client: graphql.NewClient(config.Hasura.Endpoint, httpClient), - } -} - -// InsertParsedMessage inserts a new ParsedMessage into the Hasura GraphQL API -func (hr *HasuraRepository) CreateNew(pm *domain.ParsedMessage) error { - log := utils.GetSugaredLogger() - bodyString, _ := json.Marshal(pm.BodyData) - secondAddress, _ := json.Marshal(pm.SecondaryAddresses) - var err error - msgUuid := utils.GetUuid(pm.Uuid) - variables := Aviation_telegrams_insert_input{ - Message_id: pm.MessageID, - Priority_indicator: pm.PriorityIndicator, - Primary_address: pm.PrimaryAddress, - Secondary_addresses: string(secondAddress), - Content: pm.Content, - Body_data: bodyString, - Category: pm.Category, - Date_time: pm.DateTime, - Dispatched_at: pm.DispatchedAt, - Uuid: msgUuid, - Received_at: pm.ReceivedAt, - Originator: pm.Originator, - Originator_date_time: pm.OriginatorDateTime, - } - resp, err := newMessage(context.Background(), hr.client, variables) - if err != nil { - return err - } - // fmt.Printf("Inserted new message: %v\n", resp) - log.Infof("Saved : %v\n", resp) - return nil -} diff --git a/internal/repository/hasura_test.go b/internal/repository/hasura_test.go deleted file mode 100644 index 0f1169c..0000000 --- a/internal/repository/hasura_test.go +++ /dev/null @@ -1,89 +0,0 @@ -package repository - -import ( - "testing" - - . "github.com/onsi/ginkgo/v2" - . "github.com/onsi/gomega" -) - -func TestConfig(t *testing.T) { - RegisterFailHandler(Fail) - RunSpecs(t, "Repositories Suite") -} - -var _ = Describe("Repositories", func() { - // Context("Just a simple test", func() { - - // // Define a struct for the mutation input to match the expected GraphQL input - // // Define a struct for the mutation input to match the expected GraphQL input - // type aviation_user_insert_input struct { - // Name graphql.String `json:"name"` - // Email graphql.String `json:"email"` - // UpdateAt graphql.String `json:"update_at"` // Use string for timestamp - // } - // // Get the current time in RFC3339 format - // currentTime := time.Now().Format(time.RFC3339) - - // input := aviation_user_insert_input{ - // Name: graphql.String("new"), - // Email: graphql.String("2@2.com"), - // UpdateAt: graphql.String(currentTime), // Use formatted string - // } - - // // Define the mutation - // var mutation struct { - // InsertAviationUserOne struct { - // ID int `json:"id"` - // Name string - // } `graphql:"insert_aviation_user_one(object: $object)"` - // } - - // // Define the mutation variables - // // Define the mutation variables - // variables := map[string]interface{}{ - // "object": input, - // } - // // Define the mutation - - // client := graphql.NewClient("http://localhost:8080/v1/graphql", nil) - - // err := client.Mutate(context.Background(), &mutation, variables) - // It("should not error", func() { - // Expect(err).NotTo(HaveOccurred()) - // }) - - // It("name should be new", func() { - // Expect(mutation.InsertAviationUserOne.Name).To(Equal("new")) - // }) - - // }) - - Context("Hasura Repository", func() { - // var repository *HasuraRepository - // var uuid = "uuid" - // BeforeEach(func() { - - // }) - // It("should mutate a parsed message", func() { - // repository := NewHasuraRepo("http://localhost:8080/v1/graphql", "aviation-test") - // parseMessage := &domain.ParsedMessage{ - // Uuid: uuid, - // MessageID: "message_id", - // DateTime: "date_time", - // PriorityIndicator: "priority_indicator", - // PrimaryAddress: "primary_address", - // SecondaryAddresses: []string{"secondary_addresses"}, - // Originator: "originator", - // OriginatorDateTime: "originator_date_time", - // Category: "category", - // BodyAndFooter: "body_and_footer", - // BodyData: domain.ARR{AircraftID: "aircraft_id", Category: "ARR", DepartureAirport: "departure_airport", DepartureTime: "departure_time", ArrivalAirport: "arrival_airport", ArrivalTime: "arrival_time"}, - // ReceivedAt: time.Now(), - // } - // err := repository.InsertParsedMessage(parseMessage) - // Expect(err).NotTo(HaveOccurred()) - - // }) - }) -}) diff --git a/internal/repository/schema.graphql b/internal/repository/schema.graphql deleted file mode 100644 index 8bdba6f..0000000 --- a/internal/repository/schema.graphql +++ /dev/null @@ -1,785 +0,0 @@ -schema { - query: query_root - mutation: mutation_root - subscription: subscription_root -} - -"""whether this query should be cached (Hasura Cloud only)""" -directive @cached( - """measured in seconds""" - ttl: Int! = 60 - - """refresh the cache entry""" - refresh: Boolean! = false -) on QUERY - -""" -Boolean expression to compare columns of type "Boolean". All fields are combined with logical 'AND'. -""" -input Boolean_comparison_exp { - _eq: Boolean - _gt: Boolean - _gte: Boolean - _in: [Boolean!] - _is_null: Boolean - _lt: Boolean - _lte: Boolean - _neq: Boolean - _nin: [Boolean!] -} - -""" -Boolean expression to compare columns of type "String". All fields are combined with logical 'AND'. -""" -input String_comparison_exp { - _eq: String - _gt: String - _gte: String - - """does the column match the given case-insensitive pattern""" - _ilike: String - _in: [String!] - - """ - does the column match the given POSIX regular expression, case insensitive - """ - _iregex: String - _is_null: Boolean - - """does the column match the given pattern""" - _like: String - _lt: String - _lte: String - _neq: String - - """does the column NOT match the given case-insensitive pattern""" - _nilike: String - _nin: [String!] - - """ - does the column NOT match the given POSIX regular expression, case insensitive - """ - _niregex: String - - """does the column NOT match the given pattern""" - _nlike: String - - """ - does the column NOT match the given POSIX regular expression, case sensitive - """ - _nregex: String - - """does the column NOT match the given SQL regular expression""" - _nsimilar: String - - """ - does the column match the given POSIX regular expression, case sensitive - """ - _regex: String - - """does the column match the given SQL regular expression""" - _similar: String -} - -""" -columns and relationships of "aviation.telegrams" -""" -type aviation_telegrams { - body_data( - """JSON select path""" - path: String - ): jsonb - category: String - content: String - date_time: String - dispatched_at: timestamp - message_id: String - need_dispatch: Boolean - originator: String - originator_date_time: String - parsed_at: timestamp - primary_address: String - priority_indicator: String - received_at: timestamp! - secondary_addresses: String - uuid: uuid! -} - -""" -aggregated selection of "aviation.telegrams" -""" -type aviation_telegrams_aggregate { - aggregate: aviation_telegrams_aggregate_fields - nodes: [aviation_telegrams!]! -} - -""" -aggregate fields of "aviation.telegrams" -""" -type aviation_telegrams_aggregate_fields { - count(columns: [aviation_telegrams_select_column!], distinct: Boolean): Int! - max: aviation_telegrams_max_fields - min: aviation_telegrams_min_fields -} - -"""append existing jsonb value of filtered columns with new jsonb value""" -input aviation_telegrams_append_input { - body_data: jsonb -} - -""" -Boolean expression to filter rows from the table "aviation.telegrams". All fields are combined with a logical 'AND'. -""" -input aviation_telegrams_bool_exp { - _and: [aviation_telegrams_bool_exp!] - _not: aviation_telegrams_bool_exp - _or: [aviation_telegrams_bool_exp!] - body_data: jsonb_comparison_exp - category: String_comparison_exp - content: String_comparison_exp - date_time: String_comparison_exp - dispatched_at: timestamp_comparison_exp - message_id: String_comparison_exp - need_dispatch: Boolean_comparison_exp - originator: String_comparison_exp - originator_date_time: String_comparison_exp - parsed_at: timestamp_comparison_exp - primary_address: String_comparison_exp - priority_indicator: String_comparison_exp - received_at: timestamp_comparison_exp - secondary_addresses: String_comparison_exp - uuid: uuid_comparison_exp -} - -""" -unique or primary key constraints on table "aviation.telegrams" -""" -enum aviation_telegrams_constraint { - """ - unique or primary key constraint on columns "uuid" - """ - telegrams_pkey -} - -""" -delete the field or element with specified path (for JSON arrays, negative integers count from the end) -""" -input aviation_telegrams_delete_at_path_input { - body_data: [String!] -} - -""" -delete the array element with specified index (negative integers count from the end). throws an error if top level container is not an array -""" -input aviation_telegrams_delete_elem_input { - body_data: Int -} - -""" -delete key/value pair or string element. key/value pairs are matched based on their key value -""" -input aviation_telegrams_delete_key_input { - body_data: String -} - -""" -input type for inserting data into table "aviation.telegrams" -""" -input aviation_telegrams_insert_input { - body_data: jsonb - category: String - content: String - date_time: String - dispatched_at: timestamp - message_id: String - need_dispatch: Boolean - originator: String - originator_date_time: String - parsed_at: timestamp - primary_address: String - priority_indicator: String - received_at: timestamp - secondary_addresses: String - uuid: uuid -} - -"""aggregate max on columns""" -type aviation_telegrams_max_fields { - category: String - content: String - date_time: String - dispatched_at: timestamp - message_id: String - originator: String - originator_date_time: String - parsed_at: timestamp - primary_address: String - priority_indicator: String - received_at: timestamp - secondary_addresses: String - uuid: uuid -} - -"""aggregate min on columns""" -type aviation_telegrams_min_fields { - category: String - content: String - date_time: String - dispatched_at: timestamp - message_id: String - originator: String - originator_date_time: String - parsed_at: timestamp - primary_address: String - priority_indicator: String - received_at: timestamp - secondary_addresses: String - uuid: uuid -} - -""" -response of any mutation on the table "aviation.telegrams" -""" -type aviation_telegrams_mutation_response { - """number of rows affected by the mutation""" - affected_rows: Int! - - """data from the rows affected by the mutation""" - returning: [aviation_telegrams!]! -} - -""" -on_conflict condition type for table "aviation.telegrams" -""" -input aviation_telegrams_on_conflict { - constraint: aviation_telegrams_constraint! - update_columns: [aviation_telegrams_update_column!]! = [] - where: aviation_telegrams_bool_exp -} - -"""Ordering options when selecting data from "aviation.telegrams".""" -input aviation_telegrams_order_by { - body_data: order_by - category: order_by - content: order_by - date_time: order_by - dispatched_at: order_by - message_id: order_by - need_dispatch: order_by - originator: order_by - originator_date_time: order_by - parsed_at: order_by - primary_address: order_by - priority_indicator: order_by - received_at: order_by - secondary_addresses: order_by - uuid: order_by -} - -"""primary key columns input for table: aviation.telegrams""" -input aviation_telegrams_pk_columns_input { - uuid: uuid! -} - -"""prepend existing jsonb value of filtered columns with new jsonb value""" -input aviation_telegrams_prepend_input { - body_data: jsonb -} - -""" -select columns of table "aviation.telegrams" -""" -enum aviation_telegrams_select_column { - """column name""" - body_data - - """column name""" - category - - """column name""" - content - - """column name""" - date_time - - """column name""" - dispatched_at - - """column name""" - message_id - - """column name""" - need_dispatch - - """column name""" - originator - - """column name""" - originator_date_time - - """column name""" - parsed_at - - """column name""" - primary_address - - """column name""" - priority_indicator - - """column name""" - received_at - - """column name""" - secondary_addresses - - """column name""" - uuid -} - -""" -input type for updating data in table "aviation.telegrams" -""" -input aviation_telegrams_set_input { - body_data: jsonb - category: String - content: String - date_time: String - dispatched_at: timestamp - message_id: String - need_dispatch: Boolean - originator: String - originator_date_time: String - parsed_at: timestamp - primary_address: String - priority_indicator: String - received_at: timestamp - secondary_addresses: String - uuid: uuid -} - -""" -Streaming cursor of the table "aviation_telegrams" -""" -input aviation_telegrams_stream_cursor_input { - """Stream column input with initial value""" - initial_value: aviation_telegrams_stream_cursor_value_input! - - """cursor ordering""" - ordering: cursor_ordering -} - -"""Initial value of the column from where the streaming should start""" -input aviation_telegrams_stream_cursor_value_input { - body_data: jsonb - category: String - content: String - date_time: String - dispatched_at: timestamp - message_id: String - need_dispatch: Boolean - originator: String - originator_date_time: String - parsed_at: timestamp - primary_address: String - priority_indicator: String - received_at: timestamp - secondary_addresses: String - uuid: uuid -} - -""" -update columns of table "aviation.telegrams" -""" -enum aviation_telegrams_update_column { - """column name""" - body_data - - """column name""" - category - - """column name""" - content - - """column name""" - date_time - - """column name""" - dispatched_at - - """column name""" - message_id - - """column name""" - need_dispatch - - """column name""" - originator - - """column name""" - originator_date_time - - """column name""" - parsed_at - - """column name""" - primary_address - - """column name""" - priority_indicator - - """column name""" - received_at - - """column name""" - secondary_addresses - - """column name""" - uuid -} - -input aviation_telegrams_updates { - """append existing jsonb value of filtered columns with new jsonb value""" - _append: aviation_telegrams_append_input - - """ - delete the field or element with specified path (for JSON arrays, negative integers count from the end) - """ - _delete_at_path: aviation_telegrams_delete_at_path_input - - """ - delete the array element with specified index (negative integers count from the end). throws an error if top level container is not an array - """ - _delete_elem: aviation_telegrams_delete_elem_input - - """ - delete key/value pair or string element. key/value pairs are matched based on their key value - """ - _delete_key: aviation_telegrams_delete_key_input - - """prepend existing jsonb value of filtered columns with new jsonb value""" - _prepend: aviation_telegrams_prepend_input - - """sets the columns of the filtered rows to the given values""" - _set: aviation_telegrams_set_input - - """filter the rows which have to be updated""" - where: aviation_telegrams_bool_exp! -} - -"""ordering argument of a cursor""" -enum cursor_ordering { - """ascending ordering of the cursor""" - ASC - - """descending ordering of the cursor""" - DESC -} - -scalar jsonb - -input jsonb_cast_exp { - String: String_comparison_exp -} - -""" -Boolean expression to compare columns of type "jsonb". All fields are combined with logical 'AND'. -""" -input jsonb_comparison_exp { - _cast: jsonb_cast_exp - - """is the column contained in the given json value""" - _contained_in: jsonb - - """does the column contain the given json value at the top level""" - _contains: jsonb - _eq: jsonb - _gt: jsonb - _gte: jsonb - - """does the string exist as a top-level key in the column""" - _has_key: String - - """do all of these strings exist as top-level keys in the column""" - _has_keys_all: [String!] - - """do any of these strings exist as top-level keys in the column""" - _has_keys_any: [String!] - _in: [jsonb!] - _is_null: Boolean - _lt: jsonb - _lte: jsonb - _neq: jsonb - _nin: [jsonb!] -} - -"""mutation root""" -type mutation_root { - """ - delete data from the table: "aviation.telegrams" - """ - delete_aviation_telegrams( - """filter the rows which have to be deleted""" - where: aviation_telegrams_bool_exp! - ): aviation_telegrams_mutation_response - - """ - delete single row from the table: "aviation.telegrams" - """ - delete_aviation_telegrams_by_pk(uuid: uuid!): aviation_telegrams - - """ - insert data into the table: "aviation.telegrams" - """ - insert_aviation_telegrams( - """the rows to be inserted""" - objects: [aviation_telegrams_insert_input!]! - - """upsert condition""" - on_conflict: aviation_telegrams_on_conflict - ): aviation_telegrams_mutation_response - - """ - insert a single row into the table: "aviation.telegrams" - """ - insert_aviation_telegrams_one( - """the row to be inserted""" - object: aviation_telegrams_insert_input! - - """upsert condition""" - on_conflict: aviation_telegrams_on_conflict - ): aviation_telegrams - - """ - update data of the table: "aviation.telegrams" - """ - update_aviation_telegrams( - """append existing jsonb value of filtered columns with new jsonb value""" - _append: aviation_telegrams_append_input - - """ - delete the field or element with specified path (for JSON arrays, negative integers count from the end) - """ - _delete_at_path: aviation_telegrams_delete_at_path_input - - """ - delete the array element with specified index (negative integers count from the end). throws an error if top level container is not an array - """ - _delete_elem: aviation_telegrams_delete_elem_input - - """ - delete key/value pair or string element. key/value pairs are matched based on their key value - """ - _delete_key: aviation_telegrams_delete_key_input - - """prepend existing jsonb value of filtered columns with new jsonb value""" - _prepend: aviation_telegrams_prepend_input - - """sets the columns of the filtered rows to the given values""" - _set: aviation_telegrams_set_input - - """filter the rows which have to be updated""" - where: aviation_telegrams_bool_exp! - ): aviation_telegrams_mutation_response - - """ - update single row of the table: "aviation.telegrams" - """ - update_aviation_telegrams_by_pk( - """append existing jsonb value of filtered columns with new jsonb value""" - _append: aviation_telegrams_append_input - - """ - delete the field or element with specified path (for JSON arrays, negative integers count from the end) - """ - _delete_at_path: aviation_telegrams_delete_at_path_input - - """ - delete the array element with specified index (negative integers count from the end). throws an error if top level container is not an array - """ - _delete_elem: aviation_telegrams_delete_elem_input - - """ - delete key/value pair or string element. key/value pairs are matched based on their key value - """ - _delete_key: aviation_telegrams_delete_key_input - - """prepend existing jsonb value of filtered columns with new jsonb value""" - _prepend: aviation_telegrams_prepend_input - - """sets the columns of the filtered rows to the given values""" - _set: aviation_telegrams_set_input - pk_columns: aviation_telegrams_pk_columns_input! - ): aviation_telegrams - - """ - update multiples rows of table: "aviation.telegrams" - """ - update_aviation_telegrams_many( - """updates to execute, in order""" - updates: [aviation_telegrams_updates!]! - ): [aviation_telegrams_mutation_response] -} - -"""column ordering options""" -enum order_by { - """in ascending order, nulls last""" - asc - - """in ascending order, nulls first""" - asc_nulls_first - - """in ascending order, nulls last""" - asc_nulls_last - - """in descending order, nulls first""" - desc - - """in descending order, nulls first""" - desc_nulls_first - - """in descending order, nulls last""" - desc_nulls_last -} - -type query_root { - """ - fetch data from the table: "aviation.telegrams" - """ - aviation_telegrams( - """distinct select on columns""" - distinct_on: [aviation_telegrams_select_column!] - - """limit the number of rows returned""" - limit: Int - - """skip the first n rows. Use only with order_by""" - offset: Int - - """sort the rows by one or more columns""" - order_by: [aviation_telegrams_order_by!] - - """filter the rows returned""" - where: aviation_telegrams_bool_exp - ): [aviation_telegrams!]! - - """ - fetch aggregated fields from the table: "aviation.telegrams" - """ - aviation_telegrams_aggregate( - """distinct select on columns""" - distinct_on: [aviation_telegrams_select_column!] - - """limit the number of rows returned""" - limit: Int - - """skip the first n rows. Use only with order_by""" - offset: Int - - """sort the rows by one or more columns""" - order_by: [aviation_telegrams_order_by!] - - """filter the rows returned""" - where: aviation_telegrams_bool_exp - ): aviation_telegrams_aggregate! - - """ - fetch data from the table: "aviation.telegrams" using primary key columns - """ - aviation_telegrams_by_pk(uuid: uuid!): aviation_telegrams -} - -type subscription_root { - """ - fetch data from the table: "aviation.telegrams" - """ - aviation_telegrams( - """distinct select on columns""" - distinct_on: [aviation_telegrams_select_column!] - - """limit the number of rows returned""" - limit: Int - - """skip the first n rows. Use only with order_by""" - offset: Int - - """sort the rows by one or more columns""" - order_by: [aviation_telegrams_order_by!] - - """filter the rows returned""" - where: aviation_telegrams_bool_exp - ): [aviation_telegrams!]! - - """ - fetch aggregated fields from the table: "aviation.telegrams" - """ - aviation_telegrams_aggregate( - """distinct select on columns""" - distinct_on: [aviation_telegrams_select_column!] - - """limit the number of rows returned""" - limit: Int - - """skip the first n rows. Use only with order_by""" - offset: Int - - """sort the rows by one or more columns""" - order_by: [aviation_telegrams_order_by!] - - """filter the rows returned""" - where: aviation_telegrams_bool_exp - ): aviation_telegrams_aggregate! - - """ - fetch data from the table: "aviation.telegrams" using primary key columns - """ - aviation_telegrams_by_pk(uuid: uuid!): aviation_telegrams - - """ - fetch data from the table in a streaming manner: "aviation.telegrams" - """ - aviation_telegrams_stream( - """maximum number of rows returned in a single batch""" - batch_size: Int! - - """cursor to stream the results returned by the query""" - cursor: [aviation_telegrams_stream_cursor_input]! - - """filter the rows returned""" - where: aviation_telegrams_bool_exp - ): [aviation_telegrams!]! -} - -scalar timestamp - -""" -Boolean expression to compare columns of type "timestamp". All fields are combined with logical 'AND'. -""" -input timestamp_comparison_exp { - _eq: timestamp - _gt: timestamp - _gte: timestamp - _in: [timestamp!] - _is_null: Boolean - _lt: timestamp - _lte: timestamp - _neq: timestamp - _nin: [timestamp!] -} - -scalar uuid - -""" -Boolean expression to compare columns of type "uuid". All fields are combined with logical 'AND'. -""" -input uuid_comparison_exp { - _eq: uuid - _gt: uuid - _gte: uuid - _in: [uuid!] - _is_null: Boolean - _lt: uuid - _lte: uuid - _neq: uuid - _nin: [uuid!] -} - diff --git a/main b/main new file mode 100755 index 0000000..47774dd Binary files /dev/null and b/main differ diff --git a/pkg/di/wire.go b/pkg/di/wire.go new file mode 100644 index 0000000..7082dfe --- /dev/null +++ b/pkg/di/wire.go @@ -0,0 +1,45 @@ +//go:build wireinject +// +build wireinject + +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" + + "github.com/google/wire" +) + +// InitializeApp initializes the application with all dependencies +func InitializeApp() (*app.MessageProcessor, *nats.Consumer, error) { + wire.Build( + // Config + config.ProvideConfig, + + // Logger + log.ProvideLogger, + + // Database + postgres.ProvideDB, + postgres.ProvideRepository, + + // NATS + nats.ProvideJetStream, + nats.ProvidePublisher, + + // Parser + parser.ProvideParser, + + // App + app.NewMessageProcessor, + + // Consumer + nats.ProvideConsumer, + ) + return nil, nil, nil +} + diff --git a/pkg/di/wire_gen.go b/pkg/di/wire_gen.go new file mode 100644 index 0000000..e1128c2 --- /dev/null +++ b/pkg/di/wire_gen.go @@ -0,0 +1,51 @@ +// Code generated by Wire. DO NOT EDIT. + +//go:build !wireinject +// +build !wireinject + +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" +) + +// InitializeApp initializes the application with all dependencies +func InitializeApp() (*app.MessageProcessor, *nats.Consumer, error) { + configConfig, err := config.ProvideConfig() + 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 + } + natsJetStreamContext, err := nats.ProvideJetStream(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(natsJetStreamContext, appMessageProcessor, configConfig, zapLogger) + if err != nil { + return nil, nil, err + } + return appMessageProcessor, natsConsumer, nil +} +