diff --git a/internal/nats/sub.go b/internal/nats/sub.go index b433b6c..96d63c8 100644 --- a/internal/nats/sub.go +++ b/internal/nats/sub.go @@ -3,6 +3,7 @@ package nats import ( "caatsm/internal/config" "caatsm/internal/parsers" + "caatsm/internal/repository" "context" "errors" "fmt" @@ -14,11 +15,12 @@ import ( ) type NatsHandler struct { - config *config.Config + config *config.Config + hasuraRepo *repository.HasuraRepository } func NewNatsHandler(config *config.Config) *NatsHandler { - return &NatsHandler{config: config} + return &NatsHandler{config: config, hasuraRepo: repository.NewHasuraRepo(config.Hasura.Endpoint, config.Hasura.Secret)} } func (n *NatsHandler) Subscribe() { @@ -72,7 +74,10 @@ func (n *NatsHandler) handleMessage(msg *message.Message) error { return err } else { // log.Info("message ", map[string]interface{}{"message": parsed}) - fmt.Println("message ", parsed) + fmt.Printf("message [%s]: %v\n", parsed.Uuid, parsed) + if err := n.hasuraRepo.InsertParsedMessage(parsed); err != nil { + fmt.Print("error inserting message", err) + } } return nil } diff --git a/internal/repository/hasura.go b/internal/repository/hasura.go index a812bf8..1e97c9f 100644 --- a/internal/repository/hasura.go +++ b/internal/repository/hasura.go @@ -4,6 +4,7 @@ import ( "context" "fmt" "os" + "time" "caatsm/internal/domain" @@ -26,58 +27,42 @@ func NewHasuraRepo(endpoint, secret string) *HasuraRepository { } } -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 { +func (hr *HasuraRepository) InsertParsedMessage(pm *domain.ParsedMessage) error { + if hr.hasuraClient == nil { + return fmt.Errorf("hasuraClient is nil") + } + var mutation struct { + InsertParsedMessages struct { + Returning []domain.ParsedMessage `json:"returning"` + } `graphql:"insert_messages_one(object: {message_id: $messageId, date_time: $dateTime, priority_indicator: $priorityIndicator, primary_address: $primaryAddress, secondary_addresses: $secondaryAddresses, originator: $originator, originator_date_time: $originatorDateTime, category: $category, body_and_footer: $bodyAndFooter, body_data: $bodyData, received_at: $receivedAt, parsed_at: $parsedAt, need_dispatch: $needDispatch})"` + } + + // Define the variables with their types 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, + "secondaryAddresses": graphql.String("second"), "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, + "bodyData": graphql.String("{}"), + "receivedAt": pm.ReceivedAt.Format(time.RFC3339), + "parsedAt": pm.ParsedAt.Format(time.RFC3339), + // "dispatchedAt": pm.DispatchedAt.Format(time.RFC3339), + "needDispatch": graphql.Boolean(false), } ctx := context.Background() + // Execute the mutation with the variables 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 } diff --git a/internal/repository/hasura_test.go b/internal/repository/hasura_test.go new file mode 100644 index 0000000..910ab3c --- /dev/null +++ b/internal/repository/hasura_test.go @@ -0,0 +1,45 @@ +package repository + +import ( + "caatsm/internal/domain" + "testing" + "time" + + . "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("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: nil, + ReceivedAt: time.Now(), + } + err := repository.InsertParsedMessage(parseMessage) + Expect(err).NotTo(HaveOccurred()) + + }) + }) +})