From 6b2a38b51b19cce482b4b681ff3a26047c35bc71 Mon Sep 17 00:00:00 2001 From: nutcas3 Date: Sat, 25 Apr 2026 06:50:13 +0300 Subject: [PATCH] feat: Add RabbitMQ message queue integration for profile event publishing in carrier-connector - Add github.com/rabbitmq/amqp091-go v1.11.0 dependency to go.mod - Add messageQueue parameter to OrderProfileHandlerWithRepo and DeleteProfileHandlerWithRepo - Add PublishProfileEvent calls for profile.downloaded and profile.deleted events with goroutines - Add message queue initialization in main.go with RABBITMQ_URL environment variable - Add profile-events queue declaration with error handling and fallback - Add messageQueue parameter to setupRoutes --- apps/carrier-connector/go.mod | 1 + apps/carrier-connector/go.sum | 4 ++++ .../internal/handler/profile_handlers.go | 22 ++++++++++++++++-- apps/carrier-connector/main.go | 23 ++++++++++++++++++- apps/carrier-connector/routes.go | 9 ++++---- 5 files changed, 52 insertions(+), 7 deletions(-) diff --git a/apps/carrier-connector/go.mod b/apps/carrier-connector/go.mod index 96cae33..c2ba536 100644 --- a/apps/carrier-connector/go.mod +++ b/apps/carrier-connector/go.mod @@ -6,6 +6,7 @@ require ( github.com/gin-gonic/gin v1.12.0 github.com/joho/godotenv v1.5.1 github.com/prometheus/client_golang v1.20.5 + github.com/rabbitmq/amqp091-go v1.11.0 github.com/rs/zerolog v1.35.0 gorm.io/driver/postgres v1.6.0 gorm.io/gorm v1.31.1 diff --git a/apps/carrier-connector/go.sum b/apps/carrier-connector/go.sum index 6c1700d..943e8dd 100644 --- a/apps/carrier-connector/go.sum +++ b/apps/carrier-connector/go.sum @@ -85,6 +85,8 @@ github.com/quic-go/qpack v0.6.0 h1:g7W+BMYynC1LbYLSqRt8PBg5Tgwxn214ZZR34VIOjz8= github.com/quic-go/qpack v0.6.0/go.mod h1:lUpLKChi8njB4ty2bFLX2x4gzDqXwUpaO1DP9qMDZII= github.com/quic-go/quic-go v0.59.0 h1:OLJkp1Mlm/aS7dpKgTc6cnpynnD2Xg7C1pwL6vy/SAw= github.com/quic-go/quic-go v0.59.0/go.mod h1:upnsH4Ju1YkqpLXC305eW3yDZ4NfnNbmQRCMWS58IKU= +github.com/rabbitmq/amqp091-go v1.11.0 h1:HxIctVm9Gid/Vtn706necmZ7Wj6pgGI2eqplRbEY8O8= +github.com/rabbitmq/amqp091-go v1.11.0/go.mod h1:Hy4jKW5kQART1u+JkDTF9YYOQUHXqMuhrgxOEeS7G4o= github.com/rs/zerolog v1.35.0 h1:VD0ykx7HMiMJytqINBsKcbLS+BJ4WYjz+05us+LRTdI= github.com/rs/zerolog v1.35.0/go.mod h1:EjML9kdfa/RMA7h/6z6pYmq1ykOuA8/mjWaEvGI+jcw= github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= @@ -105,6 +107,8 @@ github.com/ugorji/go/codec v1.3.1 h1:waO7eEiFDwidsBN6agj1vJQ4AG7lh2yqXyOXqhgQuyY github.com/ugorji/go/codec v1.3.1/go.mod h1:pRBVtBSKl77K30Bv8R2P+cLSGaTtex6fsA2Wjqmfxj4= go.mongodb.org/mongo-driver/v2 v2.5.0 h1:yXUhImUjjAInNcpTcAlPHiT7bIXhshCTL3jVBkF3xaE= go.mongodb.org/mongo-driver/v2 v2.5.0/go.mod h1:yOI9kBsufol30iFsl1slpdq1I0eHPzybRWdyYUs8K/0= +go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto= +go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE= go.uber.org/mock v0.6.0 h1:hyF9dfmbgIX5EfOdasqLsWD6xqpNZlXblLB/Dbnwv3Y= go.uber.org/mock v0.6.0/go.mod h1:KiVJ4BqZJaMj4svdfmHM0AUx4NJYO8ZNpPnZn1Z+BBU= golang.org/x/arch v0.22.0 h1:c/Zle32i5ttqRXjdLyyHZESLD/bB90DCU1g9l/0YBDI= diff --git a/apps/carrier-connector/internal/handler/profile_handlers.go b/apps/carrier-connector/internal/handler/profile_handlers.go index 7fce503..ddddf47 100644 --- a/apps/carrier-connector/internal/handler/profile_handlers.go +++ b/apps/carrier-connector/internal/handler/profile_handlers.go @@ -10,12 +10,13 @@ import ( "github.com/gin-gonic/gin" "github.com/nutcas3/telecom-platform/apps/carrier-connector/internal/es2" + "github.com/nutcas3/telecom-platform/apps/carrier-connector/internal/mq" "github.com/nutcas3/telecom-platform/apps/carrier-connector/internal/repository" "github.com/nutcas3/telecom-platform/apps/carrier-connector/internal/webhook" ) // OrderProfileHandlerWithRepo orders a profile via ES2+ and persists it in the repo. -func OrderProfileHandlerWithRepo(client *es2.ES2Client, repo repository.ProfileRepository, webhookClient *webhook.WebhookClient) gin.HandlerFunc { +func OrderProfileHandlerWithRepo(client *es2.ES2Client, repo repository.ProfileRepository, webhookClient *webhook.WebhookClient, messageQueue *mq.MessageQueue) gin.HandlerFunc { return func(c *gin.Context) { var order ProfileOrder if err := c.ShouldBindJSON(&order); err != nil { @@ -78,6 +79,15 @@ func OrderProfileHandlerWithRepo(client *es2.ES2Client, repo repository.ProfileR }) } + // Publish message queue event for profile download + if messageQueue != nil { + go messageQueue.PublishProfileEvent("profile.downloaded", order.ICCID, map[string]any{ + "imsi": order.IMSI, + "profileType": order.ProfileType, + "status": downloadResp.ExecutionStatus, + }) + } + c.JSON(http.StatusOK, gin.H{ "success": true, "profile": &ProfileResponse{ @@ -170,7 +180,7 @@ func ListProfilesHandlerWithRepo(repo repository.ProfileRepository) gin.HandlerF } // DeleteProfileHandlerWithRepo deletes a profile from the SM-DP+ and marks it deleted in the repo. -func DeleteProfileHandlerWithRepo(client *es2.ES2Client, repo repository.ProfileRepository, webhookClient *webhook.WebhookClient) gin.HandlerFunc { +func DeleteProfileHandlerWithRepo(client *es2.ES2Client, repo repository.ProfileRepository, webhookClient *webhook.WebhookClient, messageQueue *mq.MessageQueue) gin.HandlerFunc { return func(c *gin.Context) { profileID := c.Param("profileId") if profileID == "" { @@ -197,6 +207,14 @@ func DeleteProfileHandlerWithRepo(client *es2.ES2Client, repo repository.Profile }) } + // Publish message queue event for profile deletion + if messageQueue != nil { + go messageQueue.PublishProfileEvent("profile.deleted", profileID, map[string]any{ + "executionStatus": deleteResp.ExecutionStatus, + "statusMessage": deleteResp.StatusMessage, + }) + } + c.JSON(http.StatusOK, gin.H{ "success": true, "message": "Profile deletion requested", diff --git a/apps/carrier-connector/main.go b/apps/carrier-connector/main.go index 1a4b692..73e3c12 100644 --- a/apps/carrier-connector/main.go +++ b/apps/carrier-connector/main.go @@ -10,6 +10,7 @@ import ( "github.com/nutcas3/telecom-platform/apps/carrier-connector/internal/config" "github.com/nutcas3/telecom-platform/apps/carrier-connector/internal/es2" handler "github.com/nutcas3/telecom-platform/apps/carrier-connector/internal/handler" + "github.com/nutcas3/telecom-platform/apps/carrier-connector/internal/mq" "github.com/nutcas3/telecom-platform/apps/carrier-connector/internal/repository" "github.com/nutcas3/telecom-platform/apps/carrier-connector/internal/webhook" "github.com/rs/zerolog" @@ -62,7 +63,27 @@ func main() { webhookAPIKey := handler.GetEnv("API_SERVER_WEBHOOK_API_KEY", "") webhookClient := webhook.NewWebhookClient(webhookURL, webhookAPIKey) - setupRoutes(router, client, profileRepo, webhookClient) + // Initialize message queue for async events (optional) + var messageQueue *mq.MessageQueue + rabbitmqURL := handler.GetEnv("RABBITMQ_URL", "") + if rabbitmqURL != "" { + mq, err := mq.NewMessageQueue(rabbitmqURL) + if err != nil { + handler.Logger.Warn().Err(err).Msg("Failed to connect to RabbitMQ, continuing without message queue") + } else { + // Declare profile events queue + if err := mq.DeclareQueue("profile-events"); err != nil { + handler.Logger.Warn().Err(err).Msg("Failed to declare profile-events queue") + mq.Close() + } else { + messageQueue = mq + handler.Logger.Info().Msg("Message queue initialized") + defer messageQueue.Close() + } + } + } + + setupRoutes(router, client, profileRepo, webhookClient, messageQueue) handler.Logger.Info().Str("port", port).Msg("Carrier Connector API server starting") if err := router.Run(":" + port); err != nil { diff --git a/apps/carrier-connector/routes.go b/apps/carrier-connector/routes.go index b41a558..5ecd492 100644 --- a/apps/carrier-connector/routes.go +++ b/apps/carrier-connector/routes.go @@ -9,12 +9,13 @@ import ( "github.com/nutcas3/telecom-platform/apps/carrier-connector/internal/es2" handler "github.com/nutcas3/telecom-platform/apps/carrier-connector/internal/handler" + "github.com/nutcas3/telecom-platform/apps/carrier-connector/internal/mq" "github.com/nutcas3/telecom-platform/apps/carrier-connector/internal/repository" "github.com/nutcas3/telecom-platform/apps/carrier-connector/internal/webhook" ) -// setupRoutes registers all HTTP routes, wiring the ES2+ client, profile repo, and webhook client. -func setupRoutes(router *gin.Engine, client *es2.ES2Client, repo repository.ProfileRepository, webhookClient *webhook.WebhookClient) { +// setupRoutes registers all HTTP routes, wiring the ES2+ client, profile repo, webhook client, and message queue. +func setupRoutes(router *gin.Engine, client *es2.ES2Client, repo repository.ProfileRepository, webhookClient *webhook.WebhookClient, messageQueue *mq.MessageQueue) { api := router.Group("/api/v1") api.GET("/health", healthHandler) api.GET("/health/ready", readinessHandler(repo)) @@ -23,10 +24,10 @@ func setupRoutes(router *gin.Engine, client *es2.ES2Client, repo repository.Prof esim := api.Group("/esim") { - esim.POST("/profiles", handler.OrderProfileHandlerWithRepo(client, repo, webhookClient)) + esim.POST("/profiles", handler.OrderProfileHandlerWithRepo(client, repo, webhookClient, messageQueue)) esim.GET("/profiles", handler.ListProfilesHandlerWithRepo(repo)) esim.GET("/profiles/:profileId", handler.GetProfileHandlerWithRepo(client, repo)) - esim.DELETE("/profiles/:profileId", handler.DeleteProfileHandlerWithRepo(client, repo, webhookClient)) + esim.DELETE("/profiles/:profileId", handler.DeleteProfileHandlerWithRepo(client, repo, webhookClient, messageQueue)) } carrier := api.Group("/carrier")