diff --git a/configs/config.dev.toml b/configs/config.dev.toml index daa7635..3bcdbac 100644 --- a/configs/config.dev.toml +++ b/configs/config.dev.toml @@ -11,4 +11,8 @@ queue = "tele-queue" server = "5s" reconnect_wait = "5s" close = "10s" -ack_wait = "5s" \ No newline at end of file +ack_wait = "5s" + +[hasura] +endpoint = "http://localhost:8080/v1/graphql" +secret = "aviation-test" \ No newline at end of file diff --git a/go.sum b/go.sum index 533a7fd..6a6f9d1 100644 --- a/go.sum +++ b/go.sum @@ -19,14 +19,16 @@ github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeN github.com/google/pprof v0.0.0-20240424215950-a892ee059fd6 h1:k7nVchz72niMH6YLQNvHSdIE7iqsQxK1P41mySCvssg= github.com/google/pprof v0.0.0-20240424215950-a892ee059fd6/go.mod h1:kf6iHlnVGwgKolg33glAes7Yg/8iWP8ukqeldJSO7jw= github.com/google/uuid v1.2.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= -github.com/google/uuid v1.4.0 h1:MtMxsa51/r9yyhkyLsVeVt0B+BGQZzpQiTQ4eHZ8bc4= -github.com/google/uuid v1.4.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= +github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= +github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= github.com/hashicorp/errwrap v1.1.0 h1:OxrOeh75EUXMY8TBjag2fzXGZ40LB6IKw45YeGUDY2I= github.com/hashicorp/errwrap v1.1.0/go.mod h1:YH+1FKiLXxHSkmPseP+kNlulaMuP3n2brvKWEqk/Jc4= github.com/hashicorp/go-multierror v1.1.1 h1:H5DkEtf6CXdFp0N0Em5UCwQpXMWke8IA0+lD48awMYo= github.com/hashicorp/go-multierror v1.1.1/go.mod h1:iw975J/qwKPdAO1clOe2L8331t/9/fmwbPZ6JB6eMoM= github.com/hashicorp/hcl v1.0.0 h1:0Anlzjpi4vEasTeNFn2mLJgTSwt0+6sfsiTG8qcWGx4= github.com/hashicorp/hcl v1.0.0/go.mod h1:E5yfLk+7swimpb2L/Alb/PJmXilQ/rhwaUYs4T20WEQ= +github.com/hasura/go-graphql-client v0.12.2 h1:cYeQK/CELtvFy2jvik4kG0b5UMGngQRYWTTXQkbGHDo= +github.com/hasura/go-graphql-client v0.12.2/go.mod h1:17qYcHgGSensF/wMAHKUhtMYaRZwZa3TyD7biqH9L3k= github.com/klauspost/compress v1.17.2 h1:RlWWUY/Dr4fL8qk9YG7DTZ7PDgME2V4csBXA8L/ixi4= github.com/klauspost/compress v1.17.2/go.mod h1:ntbaceVETuRiXiv4DpjP66DpAtAGkEQskQzEyD//IeE= github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= @@ -97,6 +99,8 @@ golang.org/x/exp v0.0.0-20230905200255-921286631fa9 h1:GoHiUyI/Tp2nVkLI2mCxVkOjs golang.org/x/exp v0.0.0-20230905200255-921286631fa9/go.mod h1:S2oDrQGGwySpoQPVqRShND87VCbxmc6bL1Yd2oYrm6k= golang.org/x/net v0.25.0 h1:d/OCCoBEUq33pjydKrGQhw7IlUPI2Oylr+8qLx49kac= golang.org/x/net v0.25.0/go.mod h1:JkAGAh7GEvH74S6FOH42FLoXpXbE/aqXSrIQjXgsiwM= +golang.org/x/oauth2 v0.21.0 h1:tsimM75w1tF/uws5rbeHzIWxEqElMehnc+iW793zsZs= +golang.org/x/oauth2 v0.21.0/go.mod h1:XYTD2NtWslqkgxebSiOHnXEap4TF09sJSc7H1sXbhtI= golang.org/x/sys v0.20.0 h1:Od9JTbYCk261bKm4M/mw7AklTlFYIa0bIp9BgSm1S8Y= golang.org/x/sys v0.20.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= golang.org/x/text v0.15.0 h1:h1V/4gjBv8v9cjcR6+AR5+/cIYK5N/WAgiv4xlsEtAk= @@ -115,3 +119,5 @@ gopkg.in/natefinch/lumberjack.v2 v2.2.1/go.mod h1:YD8tP3GAjkrDg1eZH7EGmyESg/lsYs gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= +nhooyr.io/websocket v1.8.11 h1:f/qXNc2/3DpoSZkHt1DQu6rj4zGC8JmkkLkWss0MgN0= +nhooyr.io/websocket v1.8.11/go.mod h1:rN9OFWIUwuxg4fR5tELlYC04bXYowCP9GX47ivo2l+c= diff --git a/internal/config/config.go b/internal/config/config.go index eeea758..224d778 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -20,6 +20,7 @@ type Config struct { Nats NatsConfig Subscription SubscriptionConfig Timeouts TimeoutsConfig + Hasura HasuraConfig } type NatsConfig struct { @@ -50,6 +51,11 @@ type PatternConfig struct { Expression *regexp.Regexp } +type HasuraConfig struct { + Endpoint string + Secret string +} + // LoggerConfig represents the configuration for the logger. type LoggerConfig struct { ZapConfig zap.Config `json:"zapConfig"` diff --git a/internal/config/config_test.go b/internal/config/config_test.go index 033a90c..08af0ec 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -38,6 +38,10 @@ 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()) @@ -68,6 +72,9 @@ ack_wait_timeout = "5s" 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")) }) }) diff --git a/internal/domain/aviation.go b/internal/domain/aviation.go index de88302..2b3e50b 100644 --- a/internal/domain/aviation.go +++ b/internal/domain/aviation.go @@ -86,7 +86,8 @@ NeedDispatch: false. // 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'). + // StartIndicator string `json:"startIndicator"` // 电报开始标识: The start of the message indicator (e.g., 'ZCZC'). + Uuid string `json:"uuid"` MessageID string `json:"messageId"` // 信息ID: The message ID (e.g., 'TMQ1324'). DateTime string `json:"dateTime"` // 日期时间: The date and time of the message (e.g., '150631'). PriorityIndicator string `json:"priorityIndicator"` // 优先级标识: The priority level of the message (e.g., 'FF'). diff --git a/internal/nats/sub.go b/internal/nats/sub.go index 0e8236d..b433b6c 100644 --- a/internal/nats/sub.go +++ b/internal/nats/sub.go @@ -67,6 +67,7 @@ func (n *NatsHandler) handleMessage(msg *message.Message) error { payload := string(msg.Payload) if parsed, err := parsers.Parse(payload); err != nil { // log.Error("error parsing message", err, map[string]interface{}{"payload": payload}) + parsed.Uuid = msg.UUID fmt.Print("error parsing message", err) return err } else { diff --git a/internal/parsers/aviation_parser.go b/internal/parsers/aviation_parser.go index dd1e1ec..727814f 100644 --- a/internal/parsers/aviation_parser.go +++ b/internal/parsers/aviation_parser.go @@ -149,19 +149,33 @@ func createBodyData(data map[string]string) (interface{}, error) { } } +// Parse parses the raw text message and returns a ParsedMessage. // Parse parses the raw text message and returns a ParsedMessage. func Parse(rawText string) (*domain.ParsedMessage, error) { + // Parse the header of the message message, err := ParseHeader(rawText) if err != nil { return nil, err } + + // Initialize a new body parser bodyParser := NewBodyParser() + + // Parse the body and footer of the message bodyData, err := bodyParser.Parse(message.BodyAndFooter) if err != nil { - return nil, err + // Return the message with the parsed header and the error + // message.ParsedAt = time.Now() + return &message, err } + + // Set the parsed time to the current time message.ParsedAt = time.Now() + + // Assign the parsed body data to the message message.BodyData = bodyData + + // Return the fully parsed message return &message, nil } @@ -188,7 +202,7 @@ func ParseHeader(fullMessage string) (domain.ParsedMessage, error) { // fullMessage = strings.TrimSpace(fullMessage) lines := strings.Split(fullMessage, "\n") - startIndicator, messageID, dateTime, err := parseStartIndicator(lines[0]) + _, messageID, dateTime, err := parseStartIndicator(lines[0]) if err != nil { return domain.ParsedMessage{}, err } @@ -202,7 +216,7 @@ func ParseHeader(fullMessage string) (domain.ParsedMessage, error) { secondaryAddresses, originator, originatorDateTime, bodyAndFooter := parseRemainingLines(lines[2:]) return domain.ParsedMessage{ - StartIndicator: startIndicator, + // StartIndicator: startIndicator, MessageID: messageID, DateTime: dateTime, PriorityIndicator: priorityIndicator, diff --git a/internal/parsers/aviation_parser_test.go b/internal/parsers/aviation_parser_test.go index 07c746f..d17cce0 100644 --- a/internal/parsers/aviation_parser_test.go +++ b/internal/parsers/aviation_parser_test.go @@ -54,7 +54,7 @@ ALTERNATE ROUTES ADVISED) NNNN` parsedHeader, err := ParseHeader(message) Expect(err).ToNot(HaveOccurred()) - Expect(parsedHeader.StartIndicator).To(Equal("ZCZC")) + // Expect(parsedHeader.StartIndicator).To(Equal("ZCZC")) Expect(parsedHeader.MessageID).To(Equal("TAF6789")) Expect(parsedHeader.DateTime).To(Equal("160530")) Expect(parsedHeader.PriorityIndicator).To(Equal("QU")) @@ -97,7 +97,7 @@ ALL DEPARTURES/ARRIVALS EXPECTED TO BE DELAYED) NNNN` parsedHeader, err := ParseHeader(message) Expect(err).ToNot(HaveOccurred()) - Expect(parsedHeader.StartIndicator).To(Equal("ZCZC")) + // Expect(parsedHeader.StartIndicator).To(Equal("ZCZC")) Expect(parsedHeader.MessageID).To(Equal("NOTAM1122")) Expect(parsedHeader.DateTime).To(Equal("171000")) Expect(parsedHeader.PriorityIndicator).To(Equal("QU")) diff --git a/internal/repository/hasura.go b/internal/repository/hasura.go new file mode 100644 index 0000000..a812bf8 --- /dev/null +++ b/internal/repository/hasura.go @@ -0,0 +1,83 @@ +package repository + +import ( + "context" + "fmt" + "os" + + "caatsm/internal/domain" + + "github.com/hasura/go-graphql-client" + "golang.org/x/oauth2" +) + +type HasuraRepository struct { + hasuraClient *graphql.Client +} + +// NewHasuraClient creates a new HasuraClient +func NewHasuraRepo(endpoint, secret string) *HasuraRepository { + src := oauth2.StaticTokenSource( + &oauth2.Token{AccessToken: os.Getenv("GRAPHQL_TOKEN")}, + ) + httpClient := oauth2.NewClient(context.Background(), src) + return &HasuraRepository{ + hasuraClient: graphql.NewClient(endpoint, httpClient), + } +} + +var mutation struct { + InsertParsedMessages struct { + Returning []domain.ParsedMessage `json:"returning"` + } ` + graphql:"insert_parsed_messages(objects: { + startIndicator: $startIndicator, + messageId: $messageId, + dateTime: $dateTime, + priorityIndicator: $priorityIndicator, + primaryAddress: $primaryAddress, + secondaryAddresses: $secondaryAddresses, + originator: $originator, + originatorDateTime: $originatorDateTime, + category: $category, + bodyAndFooter: $bodyAndFooter, + bodyData: $bodyData, + receivedAt: $receivedAt, + parsedAt: $parsedAt, + dispatchedAt: $dispatchedAt, + needDispatch: $needDispatch + })"` +} + +// InsertParsedMessage inserts a new ParsedMessage into the Hasura GraphQL API +func (hr *HasuraRepository) InsertParsedMessage(pm domain.ParsedMessage) error { + + variables := map[string]interface{}{ + // "startIndicator": graphql.String(pm.StartIndicator), + "messageId": graphql.String(pm.MessageID), + "dateTime": graphql.String(pm.DateTime), + "priorityIndicator": graphql.String(pm.PriorityIndicator), + "primaryAddress": graphql.String(pm.PrimaryAddress), + "secondaryAddresses": pm.SecondaryAddresses, + "originator": graphql.String(pm.Originator), + "originatorDateTime": graphql.String(pm.OriginatorDateTime), + "category": graphql.String(pm.Category), + "bodyAndFooter": graphql.String(pm.BodyAndFooter), + "bodyData": pm.BodyData, + "receivedAt": pm.ReceivedAt, + "parsedAt": pm.ParsedAt, + "dispatchedAt": pm.DispatchedAt, + "needDispatch": pm.NeedDispatch, + } + + ctx := context.Background() + err := hr.hasuraClient.Mutate(ctx, &mutation, variables) + if err != nil { + return err + } + + for _, returnedMessage := range mutation.InsertParsedMessages.Returning { + fmt.Printf("Inserted ParsedMessage: %+v\n", returnedMessage) + } + return nil +}