refactor: Update nats subscription handling

The code changes in `sub.go` update the handling of NATS subscriptions. The `Subscribe` function now takes in a `config` parameter, allowing for more flexibility in configuring the subscription. Additionally, the `Subscribe` function now uses the `handler` and `marshaler` parameters to handle incoming messages and marshal/unmarshal data, respectively. These changes improve the modularity and extensibility of the code when working with NATS subscriptions.

internal/handlers/nats_handler.go

@@ -0,0 +1,70 @@
+package handlers
+
+import (
+	"caatsm/internal/config"
+	"caatsm/internal/domain"
+	"caatsm/internal/parsers"
+	"caatsm/internal/repository"
+	"caatsm/pkg/utils"
+	"errors"
+	"fmt"
+	"sync"
+
+	"github.com/ThreeDotsLabs/watermill"
+	"github.com/ThreeDotsLabs/watermill/message"
+	nc "github.com/nats-io/nats.go"
+)
+
+type NatsHandler struct {
+	mu         sync.Mutex
+	config     *config.Config
+	hasuraRepo *repository.HasuraRepository
+}
+
+func NewNatsHandler(config *config.Config) *NatsHandler {
+	return &NatsHandler{config: config, hasuraRepo: repository.NewHasuraRepo(config.Hasura.Endpoint, config.Hasura.Secret)}
+}
+
+func (n *NatsHandler) HandleMessage(msg *message.Message) error {
+	n.mu.Lock()
+	defer n.mu.Unlock()
+	log := utils.GetSugaredLogger()
+	if msg.Payload == nil {
+		log.Error("empty message")
+		return fmt.Errorf("empty message")
+	}
+	payload := string(msg.Payload)
+	var parsed *domain.ParsedMessage
+	if parsed = parsers.Parse(payload); !parsed.Parsed {
+		log.Infof("not parsed: [%s] : {%s} \n", msg.UUID, msg.Payload)
+	} else {
+		log.Infof("parsed [%s]: %v\n", msg.UUID, parsed)
+
+	}
+	n.SaveMessage(parsed, msg.UUID)
+	return nil
+}
+
+func (n *NatsHandler) SaveMessage(parsed *domain.ParsedMessage, uuid string) {
+	logger := watermill.NewStdLogger(false, false)
+	if parsed != nil {
+		parsed.Uuid = uuid
+		if err := n.hasuraRepo.CreateNew(parsed); err != nil {
+			logger.Error("error inserting message", err, map[string]interface{}{"message": parsed})
+		}
+	}
+}
+
+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
+}