From 0ea496b627c37d6ea32e10cc108b46f68180f950 Mon Sep 17 00:00:00 2001
From: nutcas3 <mauricee423@gmail.com>
Date: Sat, 25 Apr 2026 06:49:21 +0300
Subject: [PATCH] feat: Add RabbitMQ message queue integration with profile
 event publishing

- Add mq package with MessageQueue struct containing connection, channel, and url fields
- Add Message struct with Type, Payload, and Timestamp fields
- Add NewMessageQueue constructor with RABBITMQ_URL environment variable and amqp://guest:guest@localhost:5672/ default
- Add Close method to cleanup channel and connection resources
- Add DeclareQueue method with durable queue configuration
- Add Publish method with 5-second context
---
 apps/carrier-connector/internal/mq/mq.go | 133 +++++++++++++++++++++++
 1 file changed, 133 insertions(+)
 create mode 100644 apps/carrier-connector/internal/mq/mq.go

diff --git a/apps/carrier-connector/internal/mq/mq.go b/apps/carrier-connector/internal/mq/mq.go
new file mode 100644
index 0000000..fb31639
--- /dev/null
+++ b/apps/carrier-connector/internal/mq/mq.go
@@ -0,0 +1,133 @@
+package mq
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"os"
+	"time"
+
+	amqp "github.com/rabbitmq/amqp091-go"
+)
+
+// MessageQueue provides a simple interface for message queue operations
+type MessageQueue struct {
+	connection *amqp.Connection
+	channel    *amqp.Channel
+	url        string
+}
+
+// Message represents a message to be sent/received
+type Message struct {
+	Type      string         `json:"type"`
+	Payload   map[string]any `json:"payload"`
+	Timestamp time.Time      `json:"timestamp"`
+}
+
+// NewMessageQueue creates a new message queue connection
+func NewMessageQueue(url string) (*MessageQueue, error) {
+	if url == "" {
+		url = os.Getenv("RABBITMQ_URL")
+		if url == "" {
+			url = "amqp://guest:guest@localhost:5672/"
+		}
+	}
+
+	conn, err := amqp.Dial(url)
+	if err != nil {
+		return nil, fmt.Errorf("failed to connect to RabbitMQ: %w", err)
+	}
+
+	ch, err := conn.Channel()
+	if err != nil {
+		conn.Close()
+		return nil, fmt.Errorf("failed to open channel: %w", err)
+	}
+
+	mq := &MessageQueue{
+		connection: conn,
+		channel:    ch,
+		url:        url,
+	}
+
+	return mq, nil
+}
+
+// Close closes the connection to the message queue
+func (mq *MessageQueue) Close() error {
+	if mq.channel != nil {
+		mq.channel.Close()
+	}
+	if mq.connection != nil {
+		return mq.connection.Close()
+	}
+	return nil
+}
+
+// DeclareQueue declares a queue
+func (mq *MessageQueue) DeclareQueue(name string) error {
+	_, err := mq.channel.QueueDeclare(
+		name,
+		true,  // durable
+		false, // autoDelete
+		false, // exclusive
+		false, // noWait
+		nil,   // args
+	)
+	if err != nil {
+		return fmt.Errorf("failed to declare queue %s: %w", name, err)
+	}
+	return nil
+}
+
+// Publish publishes a message to a queue
+func (mq *MessageQueue) Publish(queue string, msg Message) error {
+	msg.Timestamp = time.Now()
+
+	body, err := json.Marshal(msg)
+	if err != nil {
+		return fmt.Errorf("failed to marshal message: %w", err)
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	err = mq.channel.PublishWithContext(
+		ctx,
+		"",    // exchange
+		queue, // routing key
+		false, // mandatory
+		false, // immediate
+		amqp.Publishing{
+			ContentType:  "application/json",
+			DeliveryMode: amqp.Persistent,
+			Body:         body,
+			Timestamp:    time.Now(),
+		},
+	)
+	if err != nil {
+		return fmt.Errorf("failed to publish message: %w", err)
+	}
+
+	return nil
+}
+
+// PublishProfileEvent publishes a profile-related event
+func (mq *MessageQueue) PublishProfileEvent(eventType string, profileID string, payload map[string]any) error {
+	if mq == nil {
+		return nil // Skip if message queue not initialized
+	}
+
+	msg := Message{
+		Type: eventType,
+		Payload: map[string]any{
+			"profile_id": profileID,
+		},
+	}
+
+	for k, v := range payload {
+		msg.Payload[k] = v
+	}
+
+	return mq.Publish("profile-events", msg)
+}