refactor: Improve ARR body parsing logic

internal/nats/sub.go

@@ -9,6 +9,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"sync"
 
 	"github.com/ThreeDotsLabs/watermill"
 	"github.com/ThreeDotsLabs/watermill-nats/v2/pkg/nats"
@@ -17,6 +18,7 @@ import (
 )
 
 type NatsHandler struct {
+	mu         sync.Mutex
 	config     *config.Config
 	hasuraRepo *repository.HasuraRepository
 }
@@ -63,6 +65,8 @@ func (n *NatsHandler) Subscribe() {
 	}
 }
 func (n *NatsHandler) handleMessage(msg *message.Message) error {
+	n.mu.Lock()
+	defer n.mu.Unlock()
 	log := utils.GetSugaredLogger()
 	if msg.Payload == nil {
 		// fmt.Println("empty message")
@@ -83,10 +87,12 @@ func (n *NatsHandler) handleMessage(msg *message.Message) error {
 		log.Infof("parsed [%s]: %v\n", msg.UUID, parsed)
 
 	}
+	if parsed != nil {
 		parsed.Uuid = msg.UUID
 		if err = n.hasuraRepo.CreateNew(parsed); err != nil {
 			fmt.Print("error inserting message", err)
 		}
+	}
 	return err
 }
 

internal/parsers/aviation_parser.go

@@ -3,9 +3,11 @@ package parsers
 import (
 	"caatsm/internal/config"
 	"caatsm/internal/domain"
+	"caatsm/pkg/utils"
 	"fmt"
 	"regexp"
 	"strings"
+	"sync"
 	"time"
 )
 
@@ -30,27 +32,28 @@ var (
 	otherPatterns      = []*regexp.Regexp{navPattern, remarkPattern, selPattern, pbnPattern, eetPattern, performancePattern, reroutePattern}
 )
 
+var mu sync.Mutex
+
 type BodyParser struct {
+	bodyMu       sync.Mutex
 	bodyPatterns map[string]config.BodyConfig
 }
 
-// NewBodyParser initializes a BodyParser with the default body patterns.
 func NewBodyParser() *BodyParser {
 	return &BodyParser{bodyPatterns: config.GetBodyPatterns()}
 }
 
-// GetBodyPatterns returns the body patterns used by the parser.
 func (bp *BodyParser) GetBodyPatterns() map[string]config.BodyConfig {
 	return bp.bodyPatterns
 }
 
-// SetBodyPatterns sets the body patterns for the parser.
 func (bp *BodyParser) SetBodyPatterns(patterns map[string]config.BodyConfig) {
 	bp.bodyPatterns = patterns
 }
 
-// Parse attempts to parse the body text using the configured patterns.
 func (bp *BodyParser) Parse(body string) (string, interface{}, error) {
+	bp.bodyMu.Lock()
+	defer bp.bodyMu.Unlock()
 	body = strings.TrimSpace(body)
 	category := findCategory(body)
 	if category == "" {
@@ -61,14 +64,13 @@ func (bp *BodyParser) Parse(body string) (string, interface{}, error) {
 		for _, p := range patternConfig.Patterns {
 			if match := p.Expression.FindStringSubmatch(body); match != nil {
 				data := extractData(match, p.Expression)
-				return createBodyData(data)
+				return bp.createBodyData(data)
 			}
 		}
 	}
 	return "", nil, fmt.Errorf("no matching pattern found for body: %s", body)
 }
 
-// findCategory extracts the category from the body text using regex.
 func findCategory(body string) string {
 	if match := categoryRegex.FindStringSubmatch(body); match != nil {
 		for i, name := range categoryRegex.SubexpNames() {
@@ -80,7 +82,6 @@ func findCategory(body string) string {
 	return ""
 }
 
-// extractData extracts named groups from the regex match and returns them as a map.
 func extractData(match []string, re *regexp.Regexp) map[string]string {
 	data := make(map[string]string)
 	for i, name := range re.SubexpNames() {
@@ -91,8 +92,7 @@ func extractData(match []string, re *regexp.Regexp) map[string]string {
 	return data
 }
 
-// createBodyData creates the appropriate domain object based on the type of message.
+func (bp *BodyParser) createBodyData(data map[string]string) (string, interface{}, error) {
-func createBodyData(data map[string]string) (string, interface{}, error) {
 	switch category := data["category"]; category {
 	case "ARR":
 		return category, &domain.ARR{
@@ -139,12 +139,13 @@ func createBodyData(data map[string]string) (string, interface{}, error) {
 			Remarks:                 otherData["remark"],
 		}, nil
 	default:
-		return category, nil, fmt.Errorf("cann't parse : %s", category)
+		return category, nil, fmt.Errorf("cannot parse: %s", category)
 	}
 }
 
-// Parse parses the raw text message and returns a ParsedMessage.
 func Parse(rawText string) (*domain.ParsedMessage, error) {
+	mu.Lock()
+	defer mu.Unlock()
 	message, err := ParseHeader(rawText)
 	if err != nil {
 		return nil, err
@@ -163,7 +164,6 @@ func Parse(rawText string) (*domain.ParsedMessage, error) {
 	return &message, nil
 }
 
-// clean removes empty lines from a given text and extracts the body only.
 func clean(text string) string {
 	cleanedText := emptyLineRemove.ReplaceAllString(text, "")
 	cleanText := strings.ReplaceAll(cleanedText, "\n\n", "\n")
@@ -177,11 +177,16 @@ func clean(text string) string {
 	return ""
 }
 
-// ParseHeader parses the header of the message and returns a ParsedMessage struct.
 func ParseHeader(fullMessage string) (domain.ParsedMessage, error) {
+	log := utils.GetSugaredLogger()
 	fullMessage = clean(fullMessage)
 	lines := strings.Split(fullMessage, "\n")
 
+	if len(lines) < 3 {
+		log.Warnf("invalid message format: %s", fullMessage)
+		return domain.ParsedMessage{}, fmt.Errorf("invalid message format: %s", fullMessage)
+	}
+
 	_, messageID, dateTime, err := parseStartIndicator(lines[0])
 	if err != nil {
 		return domain.ParsedMessage{}, err
@@ -203,25 +208,24 @@ func ParseHeader(fullMessage string) (domain.ParsedMessage, error) {
 	}, nil
 }
 
-// parseStartIndicator parses the start indicator line.
 func parseStartIndicator(line string) (string, string, string, error) {
 	parts := strings.Fields(line)
 	if len(parts) >= 3 && strings.HasPrefix(parts[0], StartIndicatorPrefix) {
 		return parts[0], parts[1], parts[2], nil
 	}
+	utils.GetSugaredLogger().Warnf("invalid start indicator line format: %s", line)
 	return "", "", "", fmt.Errorf("invalid start indicator line format: %s", line)
 }
 
-// parsePriorityAndPrimary parses the priority indicator and primary address line.
 func parsePriorityAndPrimary(line string) (string, string) {
 	parts := strings.Fields(line)
 	if len(parts) >= 2 {
 		return parts[0], parts[1]
 	}
+	utils.GetSugaredLogger().Warnf("invalid priority and primary address line format: %s", line)
 	return "", ""
 }
 
-// parseRemainingLines parses the remaining lines of the message.
 func parseRemainingLines(lines []string) ([]string, string, string, string) {
 	var (
 		secondaryAddresses []string
@@ -238,7 +242,6 @@ func parseRemainingLines(lines []string) ([]string, string, string, string) {
 		} else {
 			switch {
 			case line == EndHeaderMarker:
-				// End header marker, do nothing
 			case strings.HasPrefix(line, "."):
 				originatorInfo := strings.Fields(line[1:])
 				if len(originatorInfo) >= 2 {
@@ -266,16 +269,15 @@ func parseRemainingLines(lines []string) ([]string, string, string, string) {
 	return secondaryAddresses, originator, originatorDateTime, bodyAndFooter.String()
 }
 
-// getOriginator extracts originator details from a line of text.
 func getOriginator(line string) (string, string) {
 	match := originator.FindStringSubmatch(line)
 	if len(match) >= 3 {
 		return match[1], match[2]
 	}
+	utils.GetSugaredLogger().Warnf("invalid originator line format: %s", line)
 	return "", ""
 }
 
-// parseOther parses additional information from the message body.
 func parseOther(text string) map[string]string {
 	data := make(map[string]string)
 	for _, re := range otherPatterns {

internal/parsers/aviation_parser_test.go

@@ -10,26 +10,19 @@ import (
 var _ = Describe("Aviation Parser", func() {
 	Describe("ParseHeader", func() {
 
-		Context("with a real arr context", func() {
+		Context("with a real ARR context", func() {
 			message := `ZCZC TMQ2530 141614
-
-
 GG ZBTJZXZX
-
-
 141614 ZSHCZTZX
-
 (ARR-CES5470-ZBTJ-ZSHC1614)
-
 NNNN`
-			It("get a clean body text", func() {
+			It("should get a clean body text", func() {
 				body := clean(message)
-				expexted := `ZCZC TMQ2530 141614
+				expected := `ZCZC TMQ2530 141614
 GG ZBTJZXZX
 141614 ZSHCZTZX
 (ARR-CES5470-ZBTJ-ZSHC1614)`
-				// fmt.Printf("\n%v \n%v\n", []byte(body), []byte(expexted))
+				Expect(body).To(Equal(expected))
-				Expect(body).To(Equal(expexted))
 			})
 		})
 
@@ -54,13 +47,11 @@ ALTERNATE ROUTES ADVISED)
 NNNN`
 			parsedHeader, err := ParseHeader(message)
 			Expect(err).ToNot(HaveOccurred())
-			// Expect(parsedHeader.StartIndicator).To(Equal("ZCZC"))
 			Expect(parsedHeader.MessageID).To(Equal("TAF6789"))
 			Expect(parsedHeader.DateTime).To(Equal("160530"))
 			Expect(parsedHeader.PriorityIndicator).To(Equal("QU"))
 			Expect(parsedHeader.PrimaryAddress).To(Equal("TSNZPCA"))
 			Expect(parsedHeader.SecondaryAddresses).To(Equal([]string{"QU PEKUDCA TSNUOCA TSNZPCA TSNUFCA"}))
-
 		})
 
 		It("should parse the header correctly with originator information", func() {
@@ -97,7 +88,6 @@ ALL DEPARTURES/ARRIVALS EXPECTED TO BE DELAYED)
 NNNN`
 			parsedHeader, err := ParseHeader(message)
 			Expect(err).ToNot(HaveOccurred())
-			// Expect(parsedHeader.StartIndicator).To(Equal("ZCZC"))
 			Expect(parsedHeader.MessageID).To(Equal("NOTAM1122"))
 			Expect(parsedHeader.DateTime).To(Equal("171000"))
 			Expect(parsedHeader.PriorityIndicator).To(Equal("QU"))
@@ -210,7 +200,6 @@ NNNN`
 				Expect(fplMessage.DestinationAndTotalTime).To(Equal("ZBAA0153"))
 				Expect(fplMessage.AlternateAirport).To(Equal("ZBYN"))
 				Expect(fplMessage.OtherInfo).To(Equal("PBN/A1B2B3B4B5D1L1 NAV/ABAS REG/B6513 EET/ZBPE0112 SEL/KMAL PER/C RIF/FRT N640 ZBYN RMK/TCAS EQUIPPED"))
-				// Expect(fplMessage.SupplementaryInfo).To(Equal("RMK/TCAS EQUIPPED"))
 				Expect(fplMessage.PBN).To(Equal("A1B2B3B4B5D1L1"))
 				Expect(fplMessage.EstimatedElapsedTime).To(Equal("ZBPE0112"))
 				Expect(fplMessage.SELCALCode).To(Equal("KMAL"))
@@ -219,29 +208,21 @@ NNNN`
 				Expect(fplMessage.Remarks).To(Equal("TCAS EQUIPPED"))
 			})
 		})
+
 	})
 
 	Describe("Parse whole real message", func() {
 
-		Context("with a real arr message", func() {
+		Context("with a real ARR message", func() {
 			message := `
 ZCZC TMQ2526 141605
 
-
 FF ZBTJZPZX
 
-
 141604 ZBACZQZX
 
-
 (ARR-JAE7433/A0132-RKSI-ZBTJ1604)
 
-
-
-
-
-
-
 NNNN
 `
 			It("should parse the whole message correctly", func() {
@@ -265,77 +246,5 @@ NNNN
 				Expect(arrmsg.ArrivalTime).To(Equal("1604"))
 			})
 		})
-
-		Context("real fpl", func() {
-			message := `ZCZC TMQ2544 141652
-
-
-FF ZBTJZXZX
-
-
-141652 ZBTJZPZX
-
-
-(FPL-JAE7433-IS
-
-
--B744/H-SXIRPZJWY/S
-
-
--ZBTJ1755
-
-
--K0926S0920 CG A326 VYK W80 HUR B339 GM A575 MANSA/K0919S0980 A575
-
-
- INTIK/K0917S0960 A575 UDA DCT BULAG A200 HATGA/K0900S1060 A308
-
-
- LARNA DCT RATKO A307 KUMOD R497 TODES B228 ZJ R22 KTL R30 SPB
-
-
- B141 RANVA/N0485F360 UP863 DEREX UP739 KOLJA UN746 GORPI UZ80
-
-
- TILAV UL87 TADUV T173 GED GED2W
- 
-
--EDDF0948 EDDK
-
-
--EET/ZMUB0100 UNKL0236 UNWW0332 UNNT0332 USRR0447 USHH0507
-
-
- USSS0535 UUYY0602 ULKK0634 ULWW0653 ULLL0720 EETT0748 EVRR0815
-
-
- ESAA0821 EPWW0848 EDUU0900
-
-
- REG/B2422 SEL/JLAD OPR/JADE CARGO DAT/S RVR/200
-
-
- NAV/RNAV1 RNAV5 RNP4
-
-
- RMK/AGCS EQUIPPED
-
-
- ACARS EQUIPPED/TCAS EQUIPPED/FOREIGN PILOT
-
-
- E/1148 P/TBN R/UV S/M J/LF D/1 15 C YELLOW
-
-
- A/WHITE GREEN)
-
-NNNN
-`
-			It("should parse the whole message correctly", func() {
-				parsedMessage, err := Parse(message)
-				Expect(err).ToNot(HaveOccurred())
-				Expect(parsedMessage).ToNot(BeNil())
-			})
-		})
 	})
 })

pkg/utils/log.go

@@ -46,15 +46,13 @@ func load() {
 		env := getEnv()
 		fmt.Printf("Enviroment : %s\n", env)
 		configFile := getConfigFile(env)
-		// fmt.Printf("Config File : %s\n", configFile)
+
-		// if err != nil {
-		// 	fmt.Printf("Error finding config file: %v\n", err)
-		// 	return
-		// }
-		// fmt.Printf("Loading config from file: %s\n", configFile)
 		config, err := loadConfig(configFile)
 		if err != nil {
 			fmt.Printf("Error loading config: %v\n", err)
+			//create a default logger
+			log, _ = zap.NewDevelopment()
+			sugar = log.Sugar()
 			return
 		}