telecom-platform/apps/carrier-connector/internal/mq/mq.go

package mq

import (
	"context"
	"encoding/json"
	"fmt"
	"maps"
	"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,
		},
	}

	maps.Copy(msg.Payload, payload)

	return mq.Publish("profile-events", msg)
}