From d78859ea07e846512ac80d8fe1b8f9ed240959c8 Mon Sep 17 00:00:00 2001 From: Lucas Monteiro Date: Tue, 28 Apr 2026 13:54:32 -0300 Subject: [PATCH 1/4] feat: ingest iOS push notifications via FCM Kafka topics Add ingestion for the new push-_ios-(single|massive) topics so the GCM pusher can consume iOS messages alongside the existing Android ones. - Extend topic regex in config and common.go to match the ios platform - Add Platform field to KafkaMessage and populate it from the topic - Subscribe pusher gcm to ios single/massive topics for every gcm app - Cover new parsing and consumer behavior with unit tests --- config/default.yaml | 2 +- config/test.yaml | 2 +- extensions/common.go | 2 +- extensions/common_test.go | 44 +++++++++++++++++++++++++++++++ extensions/kafka_consumer.go | 8 +++--- extensions/kafka_consumer_test.go | 38 ++++++++++++++++++++++++++ interfaces/queue.go | 7 ++--- pusher/gcm.go | 16 ++++++----- 8 files changed, 103 insertions(+), 16 deletions(-) diff --git a/config/default.yaml b/config/default.yaml index 097db795..3dec5dfe 100644 --- a/config/default.yaml +++ b/config/default.yaml @@ -34,7 +34,7 @@ gcm: concurrentWorkers: 10 queue: topics: - - "^push-[^-_]+_(apns|gcm)[_-](single|massive)" + - "^push-[^-_]+_(apns|gcm|ios)[_-](single|massive)" brokers: "localhost:9941" group: testGroup sessionTimeout: 6000 diff --git a/config/test.yaml b/config/test.yaml index 1e505208..c87d5a92 100644 --- a/config/test.yaml +++ b/config/test.yaml @@ -28,7 +28,7 @@ gcm: concurrentWorkers: 10 queue: topics: - - "^push-[^-_]+_(apns|gcm)[_-](single|massive)" + - "^push-[^-_]+_(apns|gcm|ios)[_-](single|massive)" brokers: "localhost:9941" group: testGroup sessionTimeout: 6000 diff --git a/extensions/common.go b/extensions/common.go index 111cfb41..cf6ab43c 100644 --- a/extensions/common.go +++ b/extensions/common.go @@ -31,7 +31,7 @@ import ( "github.com/topfreegames/pusher/interfaces" ) -var topicRegex = regexp.MustCompile("^push-([\\w]+(?:[_-][\\w]+)*)[-_](gcm|apns)") +var topicRegex = regexp.MustCompile("^push-([\\w]+(?:[_-][\\w]+)*)[-_](gcm|apns|ios)") // ParsedTopic contains game and platform extracted from topic name type ParsedTopic struct { diff --git a/extensions/common_test.go b/extensions/common_test.go index 4c3ab381..7ffacefa 100644 --- a/extensions/common_test.go +++ b/extensions/common_test.go @@ -68,5 +68,49 @@ var _ = Describe("Common", func() { Expect(err.Error()).To(Equal("json: unsupported type: chan int")) }) }) + + Describe("GetGameAndPlatformFromTopic", func() { + It("should parse gcm single topic", func() { + parsed := GetGameAndPlatformFromTopic("push-mygame_gcm-single") + Expect(parsed.Game).To(Equal("mygame")) + Expect(parsed.Platform).To(Equal("gcm")) + }) + + It("should parse gcm massive topic", func() { + parsed := GetGameAndPlatformFromTopic("push-mygame_gcm-massive") + Expect(parsed.Game).To(Equal("mygame")) + Expect(parsed.Platform).To(Equal("gcm")) + }) + + It("should parse apns single topic", func() { + parsed := GetGameAndPlatformFromTopic("push-mygame_apns-single") + Expect(parsed.Game).To(Equal("mygame")) + Expect(parsed.Platform).To(Equal("apns")) + }) + + It("should parse apns massive topic", func() { + parsed := GetGameAndPlatformFromTopic("push-mygame_apns-massive") + Expect(parsed.Game).To(Equal("mygame")) + Expect(parsed.Platform).To(Equal("apns")) + }) + + It("should parse ios single topic", func() { + parsed := GetGameAndPlatformFromTopic("push-mygame_ios-single") + Expect(parsed.Game).To(Equal("mygame")) + Expect(parsed.Platform).To(Equal("ios")) + }) + + It("should parse ios massive topic", func() { + parsed := GetGameAndPlatformFromTopic("push-mygame_ios-massive") + Expect(parsed.Game).To(Equal("mygame")) + Expect(parsed.Platform).To(Equal("ios")) + }) + + It("should parse ios topic with compound game name", func() { + parsed := GetGameAndPlatformFromTopic("push-com_my_game_ios-single") + Expect(parsed.Game).To(Equal("com_my_game")) + Expect(parsed.Platform).To(Equal("ios")) + }) + }) }) }) diff --git a/extensions/kafka_consumer.go b/extensions/kafka_consumer.go index 2014f4ca..91b7b410 100644 --- a/extensions/kafka_consumer.go +++ b/extensions/kafka_consumer.go @@ -273,10 +273,12 @@ func (q *KafkaConsumer) receiveMessage(topicPartition kafka.TopicPartition, valu q.pendingMessagesWG.Add(1) } + parsed := GetGameAndPlatformFromTopic(*topicPartition.Topic) message := interfaces.KafkaMessage{ - Game: GetGameAndPlatformFromTopic(*topicPartition.Topic).Game, - Topic: *topicPartition.Topic, - Value: value, + Game: parsed.Game, + Platform: parsed.Platform, + Topic: *topicPartition.Topic, + Value: value, } q.msgChan <- message diff --git a/extensions/kafka_consumer_test.go b/extensions/kafka_consumer_test.go index 2e1747f6..cbe58e50 100644 --- a/extensions/kafka_consumer_test.go +++ b/extensions/kafka_consumer_test.go @@ -116,6 +116,44 @@ var _ = Describe("Kafka Extension", func() { Value: val, })) }) + + It("should populate Game and Platform on received message for ios topic", func() { + topic := "push-mygame_ios-single" + startConsuming() + defer consumer.StopConsuming() + part := kafka.TopicPartition{ + Topic: &topic, + Partition: 1, + } + val := []byte("test") + event := &kafka.Message{TopicPartition: part, Value: val} + + publishEvent(event) + var received interfaces.KafkaMessage + Eventually(consumer.msgChan, 5).Should(Receive(&received)) + Expect(received.Topic).To(Equal(topic)) + Expect(received.Game).To(Equal("mygame")) + Expect(received.Platform).To(Equal("ios")) + Expect(received.Value).To(Equal(val)) + }) + + It("should populate Game and Platform on received message for gcm topic", func() { + topic := "push-mygame_gcm-massive" + startConsuming() + defer consumer.StopConsuming() + part := kafka.TopicPartition{ + Topic: &topic, + Partition: 1, + } + val := []byte("test") + event := &kafka.Message{TopicPartition: part, Value: val} + + publishEvent(event) + var received interfaces.KafkaMessage + Eventually(consumer.msgChan, 5).Should(Receive(&received)) + Expect(received.Game).To(Equal("mygame")) + Expect(received.Platform).To(Equal("gcm")) + }) }) Describe("Configuration Defaults", func() { diff --git a/interfaces/queue.go b/interfaces/queue.go index 4695d1ea..7e8243f6 100644 --- a/interfaces/queue.go +++ b/interfaces/queue.go @@ -29,9 +29,10 @@ import ( // KafkaMessage sent through the Channel. type KafkaMessage struct { - Game string - Topic string - Value []byte + Game string + Platform string + Topic string + Value []byte } // Queue interface for making new queues pluggable easily. diff --git a/pusher/gcm.go b/pusher/gcm.go index 288cb691..b0639035 100644 --- a/pusher/gcm.go +++ b/pusher/gcm.go @@ -82,13 +82,15 @@ func NewGCMPusher( } g.Queue = q for _, a := range g.Config.GetGcmAppsArray() { - singleTopic := fmt.Sprintf("push-%s_gcm-single", a) - if !slices.Contains(q.Topics, singleTopic) { - q.Topics = append(q.Topics, singleTopic) - } - massiveTopic := fmt.Sprintf("push-%s_gcm-massive", a) - if !slices.Contains(q.Topics, massiveTopic) { - q.Topics = append(q.Topics, massiveTopic) + for _, platform := range []string{"gcm", "ios"} { + singleTopic := fmt.Sprintf("push-%s_%s-single", a, platform) + if !slices.Contains(q.Topics, singleTopic) { + q.Topics = append(q.Topics, singleTopic) + } + massiveTopic := fmt.Sprintf("push-%s_%s-massive", a, platform) + if !slices.Contains(q.Topics, massiveTopic) { + q.Topics = append(q.Topics, massiveTopic) + } } } From 526880f087aa6a2b208baf3a80207034fa3ef538 Mon Sep 17 00:00:00 2001 From: Lucas Monteiro Date: Thu, 30 Apr 2026 11:22:31 -0300 Subject: [PATCH 2/4] feat(firebase): thread platform through message handler MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds a Platform field to interfaces.Message and resolves the platform ("gcm" or "ios", with empty string falling back to "gcm") from KafkaMessage.Platform at the top of HandleMessages. The resolved value is threaded through dedup, rate limiting, the duplicate-detected stat, sendPush, HandleResponses, and every feedback/stats reporter call — replacing the hardcoded "gcm" literals so iOS messages are tagged correctly across observability paths. The Firebase client still produces Android-style payloads, so this is non-breaking until upstream services start publishing on _ios-* topics. --- extensions/firebase/message_handler.go | 48 +-- extensions/firebase/message_handler_test.go | 315 ++++++++++++++++++++ interfaces/client.go | 1 + 3 files changed, 343 insertions(+), 21 deletions(-) diff --git a/extensions/firebase/message_handler.go b/extensions/firebase/message_handler.go index 7447a34e..78322f25 100644 --- a/extensions/firebase/message_handler.go +++ b/extensions/firebase/message_handler.go @@ -92,6 +92,12 @@ func (h *messageHandler) HandleMessages(ctx context.Context, msg interfaces.Kafk return } + platform := msg.Platform + if platform == "" { + platform = "gcm" + } + km.Message.Platform = platform + if km.PushExpiry > 0 && km.PushExpiry < extensions.MakeTimestamp() { l.Warnf("ignoring push message because it has expired: %s", km.Data) h.waitGroupDone() @@ -101,22 +107,22 @@ func (h *messageHandler) HandleMessages(ctx context.Context, msg interfaces.Kafk // if there is any error on deduplication, it does not block the message. dedupMsg, err := h.createDedupContentFromPayload(km) if err == nil { - uniqueMessage := h.dedup.IsUnique(ctx, km.To, dedupMsg, h.app, "gcm") + uniqueMessage := h.dedup.IsUnique(ctx, km.To, dedupMsg, h.app, platform) if !uniqueMessage { l.WithFields(logrus.Fields{ "extension": "dedup", "game": h.app, }).Debug("duplicate message detected") - extensions.StatsReporterDuplicateMessageDetected(h.statsReporters, h.app, "gcm") + extensions.StatsReporterDuplicateMessageDetected(h.statsReporters, h.app, platform) //does not return because we don't want to block the message } } else { l.WithError(err).Error("error creating deduplication content from payload") } - allowed := h.rateLimiter.Allow(ctx, km.To, msg.Game, "gcm") + allowed := h.rateLimiter.Allow(ctx, km.To, msg.Game, platform) if !allowed { - h.reportRateLimitReached(msg.Game) + h.reportRateLimitReached(msg.Game, platform) h.waitGroupDone() l.WithField("message", msg).Warn("rate limit reached") return @@ -134,7 +140,7 @@ func (h *messageHandler) HandleMessages(ctx context.Context, msg interfaces.Kafk } } before := time.Now() - defer h.reportLatency(time.Since(before)) + defer h.reportLatency(time.Since(before), platform) h.sendPush(ctx, km.Message, msg.Topic) } @@ -169,7 +175,7 @@ func (h *messageHandler) sendPush(ctx context.Context, msg interfaces.Message, t err := h.client.SendPush(ctx, msg) h.reportFirebaseLatency(time.Since(before)) - h.handleNotificationSent(topic) + h.handleNotificationSent(topic, msg.Platform) h.responsesChannel <- struct { msg interfaces.Message @@ -189,9 +195,9 @@ func (h *messageHandler) HandleResponses() { for { response := <-h.responsesChannel if response.error != nil { - h.handleNotificationFailure(response.msg, response.error) + h.handleNotificationFailure(response.msg, response.msg.Platform, response.error) } else { - h.handleNotificationAck() + h.handleNotificationAck(response.msg.Platform) } h.waitGroupDone() } @@ -199,35 +205,35 @@ func (h *messageHandler) HandleResponses() { } } -func (h *messageHandler) sendToFeedbackReporters(res interface{}) error { +func (h *messageHandler) sendToFeedbackReporters(res interface{}, platform string) error { jsonRes, err := json.Marshal(res) if err != nil { return err } for _, feedbackReporter := range h.feedbackReporters { - feedbackReporter.SendFeedback(h.app, "gcm", jsonRes) + feedbackReporter.SendFeedback(h.app, platform, jsonRes) } return nil } -func (h *messageHandler) handleNotificationSent(topic string) { +func (h *messageHandler) handleNotificationSent(topic, platform string) { for _, statsReporter := range h.statsReporters { - statsReporter.HandleNotificationSent(h.app, "gcm", topic) + statsReporter.HandleNotificationSent(h.app, platform, topic) } } -func (h *messageHandler) handleNotificationAck() { +func (h *messageHandler) handleNotificationAck(platform string) { for _, statsReporter := range h.statsReporters { - statsReporter.HandleNotificationSuccess(h.app, "gcm") + statsReporter.HandleNotificationSuccess(h.app, platform) } } -func (h *messageHandler) handleNotificationFailure(message interfaces.Message, err error) { +func (h *messageHandler) handleNotificationFailure(message interfaces.Message, platform string, err error) { pushError := translateToPushError(err) for _, statsReporter := range h.statsReporters { - statsReporter.HandleNotificationFailure(h.app, "gcm", pushError) + statsReporter.HandleNotificationFailure(h.app, platform, pushError) } for _, feedbackReporter := range h.feedbackReporters { feedback := &FeedbackResponse{ @@ -236,13 +242,13 @@ func (h *messageHandler) handleNotificationFailure(message interfaces.Message, e From: message.To, } b, _ := json.Marshal(feedback) - feedbackReporter.SendFeedback(h.app, "gcm", b) + feedbackReporter.SendFeedback(h.app, platform, b) } } -func (h *messageHandler) reportLatency(latency time.Duration) { +func (h *messageHandler) reportLatency(latency time.Duration, platform string) { for _, statsReporter := range h.statsReporters { - statsReporter.ReportSendNotificationLatency(latency, h.app, "gcm", "client", "fcm") + statsReporter.ReportSendNotificationLatency(latency, h.app, platform, "client", "fcm") } } @@ -252,9 +258,9 @@ func (h *messageHandler) reportFirebaseLatency(latency time.Duration) { } } -func (h *messageHandler) reportRateLimitReached(game string) { +func (h *messageHandler) reportRateLimitReached(game, platform string) { for _, statsReporter := range h.statsReporters { - statsReporter.NotificationRateLimitReached(game, "gcm") + statsReporter.NotificationRateLimitReached(game, platform) } } diff --git a/extensions/firebase/message_handler_test.go b/extensions/firebase/message_handler_test.go index 9da38115..8e944725 100644 --- a/extensions/firebase/message_handler_test.go +++ b/extensions/firebase/message_handler_test.go @@ -280,6 +280,321 @@ func (s *MessageHandlerTestSuite) TestHandleMessage() { s.waitGroup.Wait() }) + s.Run("should propagate ios platform through dedup, rate limiter, and stats reporters", func() { + token := uuid.NewString() + msgValue := kafkaFCMMessage{ + Message: interfaces.Message{ + To: token, + Data: map[string]interface{}{ + "title": "notification", + "body": "bodyIOS", + }, + }, + Metadata: map[string]interface{}{ + "some": "metadata", + }, + } + bytes, err := json.Marshal(msgValue) + s.Require().NoError(err) + msg := interfaces.KafkaMessage{ + Value: bytes, + Topic: "push-game_ios-single", + Game: s.game, + Platform: "ios", + } + + dedupMsg, err := createDedupContentForTest(msgValue) + s.Require().NoError(err) + + s.mockDedup.EXPECT(). + IsUnique(gomock.Any(), token, dedupMsg, s.game, "ios"). + Return(true) + + s.mockRateLimiter.EXPECT(). + Allow(gomock.Any(), token, s.game, "ios"). + Return(true) + + done := make(chan struct{}) + + s.mockClient.EXPECT(). + SendPush(gomock.Any(), gomock.Any()). + Do(func(_ context.Context, m interfaces.Message) { + s.Equal(token, m.To) + s.Equal("ios", m.Platform) + }) + + s.mockStatsReporter.EXPECT(). + ReportSendNotificationLatency(gomock.Any(), s.game, "ios", gomock.Any()).Return() + + s.mockStatsReporter.EXPECT(). + ReportFirebaseLatency(gomock.Any(), s.game, gomock.Any()).Return() + + s.mockStatsReporter.EXPECT(). + HandleNotificationSent(s.game, "ios", "push-game_ios-single"). + Do(func(game, platform, topic string) { + done <- struct{}{} + }) + + s.mockStatsReporter.EXPECT(). + HandleNotificationSuccess(s.game, "ios"). + Return() + + go s.handler.HandleResponses() + s.waitGroup.Add(1) + s.handler.HandleMessages(context.Background(), msg) + + timeout := time.NewTimer(5 * time.Second) + select { + case <-done: + case <-timeout.C: + s.Fail("timed out waiting for ios message to be processed") + } + s.waitGroup.Wait() + }) + + s.Run("should report ios platform on duplicate detected", func() { + token := uuid.NewString() + msgValue := kafkaFCMMessage{ + Message: interfaces.Message{ + To: token, + Data: map[string]interface{}{ + "title": "notification", + "body": "bodyIOSDedup", + }, + }, + } + bytes, err := json.Marshal(msgValue) + s.Require().NoError(err) + msg := interfaces.KafkaMessage{ + Value: bytes, + Topic: "push-game_ios-single", + Game: s.game, + Platform: "ios", + } + + dedupMsg, err := createDedupContentForTest(msgValue) + s.Require().NoError(err) + + s.mockDedup.EXPECT(). + IsUnique(gomock.Any(), token, dedupMsg, s.game, "ios"). + Return(false) + + s.mockStatsReporter.EXPECT(). + ReportMetricCount("duplicated_messages", int64(1), s.game, "ios"). + Return() + + s.mockRateLimiter.EXPECT(). + Allow(gomock.Any(), token, s.game, "ios"). + Return(true) + + done := make(chan struct{}) + + s.mockClient.EXPECT(). + SendPush(gomock.Any(), gomock.Any()). + Do(func(_ context.Context, m interfaces.Message) { + s.Equal(token, m.To) + done <- struct{}{} + }) + + s.mockStatsReporter.EXPECT(). + ReportSendNotificationLatency(gomock.Any(), s.game, "ios", gomock.Any()).Return() + + s.mockStatsReporter.EXPECT(). + ReportFirebaseLatency(gomock.Any(), s.game, gomock.Any()).Return() + + s.mockStatsReporter.EXPECT(). + HandleNotificationSent(s.game, "ios", "push-game_ios-single"). + Return() + + s.handler.HandleMessages(context.Background(), msg) + timeout := time.NewTimer(50 * time.Millisecond) + select { + case <-done: + case <-timeout.C: + s.Fail("timed out waiting for ios dup-detected message to be sent") + } + }) + + s.Run("should report ios platform on rate limit reached", func() { + token := uuid.NewString() + msgValue := kafkaFCMMessage{ + Message: interfaces.Message{ + To: token, + Data: map[string]interface{}{ + "title": "notification", + "body": "bodyIOSRateLimit", + }, + }, + } + bytes, err := json.Marshal(msgValue) + s.Require().NoError(err) + msg := interfaces.KafkaMessage{ + Value: bytes, + Topic: "push-game_ios-single", + Game: s.game, + Platform: "ios", + } + + dedupMsg, err := createDedupContentForTest(msgValue) + s.Require().NoError(err) + + s.mockDedup.EXPECT(). + IsUnique(gomock.Any(), token, dedupMsg, s.game, "ios"). + Return(true) + + s.mockRateLimiter.EXPECT(). + Allow(gomock.Any(), token, s.game, "ios"). + Return(false) + + s.mockStatsReporter.EXPECT(). + NotificationRateLimitReached(s.game, "ios"). + Return() + + s.waitGroup.Add(1) + s.handler.HandleMessages(context.Background(), msg) + waitWG(s.T(), s.waitGroup) + }) + + s.Run("should send ios feedback on failure", func() { + token := uuid.NewString() + msgValue := kafkaFCMMessage{ + Message: interfaces.Message{ + To: token, + Data: map[string]interface{}{ + "title": "notification", + "body": "bodyIOSFailure", + }, + }, + } + bytes, err := json.Marshal(msgValue) + s.Require().NoError(err) + msg := interfaces.KafkaMessage{ + Value: bytes, + Topic: "push-game_ios-single", + Game: s.game, + Platform: "ios", + } + + dedupMsg, err := createDedupContentForTest(msgValue) + s.Require().NoError(err) + + s.mockDedup.EXPECT(). + IsUnique(gomock.Any(), token, dedupMsg, s.game, "ios"). + Return(true) + + s.mockRateLimiter.EXPECT(). + Allow(gomock.Any(), token, s.game, "ios"). + Return(true) + + done := make(chan struct{}) + + s.mockClient.EXPECT(). + SendPush(gomock.Any(), gomock.Any()). + Return(errors.NewPushError("DEVICE_UNREGISTERED", "device unregistered")) + + s.mockStatsReporter.EXPECT(). + ReportSendNotificationLatency(gomock.Any(), s.game, "ios", gomock.Any()).Return() + + s.mockStatsReporter.EXPECT(). + ReportFirebaseLatency(gomock.Any(), s.game, gomock.Any()).Return() + + s.mockStatsReporter.EXPECT(). + HandleNotificationSent(s.game, "ios", "push-game_ios-single"). + Return() + + s.mockStatsReporter.EXPECT(). + HandleNotificationFailure(s.game, "ios", gomock.Any()) + + s.mockFeedbackReporter.EXPECT(). + SendFeedback(s.game, "ios", gomock.Any()). + DoAndReturn(func(game, platform string, feedback []byte) { + obj := &FeedbackResponse{} + err := json.Unmarshal(feedback, obj) + s.NoError(err) + s.Equal(token, obj.From) + done <- struct{}{} + }) + + go s.handler.HandleResponses() + s.waitGroup.Add(1) + s.handler.HandleMessages(context.Background(), msg) + + timeout := time.NewTimer(50 * time.Millisecond) + select { + case <-done: + case <-timeout.C: + s.Fail("timed out waiting for ios failure feedback") + } + }) + + s.Run("should default empty platform to gcm for backward compatibility", func() { + token := uuid.NewString() + msgValue := kafkaFCMMessage{ + Message: interfaces.Message{ + To: token, + Data: map[string]interface{}{ + "title": "notification", + "body": "bodyEmptyPlatform", + }, + }, + } + bytes, err := json.Marshal(msgValue) + s.Require().NoError(err) + // Platform is intentionally left empty. + msg := interfaces.KafkaMessage{ + Value: bytes, + Topic: "push-game_gcm-single", + Game: s.game, + } + + dedupMsg, err := createDedupContentForTest(msgValue) + s.Require().NoError(err) + + s.mockDedup.EXPECT(). + IsUnique(gomock.Any(), token, dedupMsg, s.game, "gcm"). + Return(true) + + s.mockRateLimiter.EXPECT(). + Allow(gomock.Any(), token, s.game, "gcm"). + Return(true) + + done := make(chan struct{}) + + s.mockClient.EXPECT(). + SendPush(gomock.Any(), gomock.Any()). + Do(func(_ context.Context, m interfaces.Message) { + s.Equal("gcm", m.Platform) + }) + + s.mockStatsReporter.EXPECT(). + ReportSendNotificationLatency(gomock.Any(), s.game, "gcm", gomock.Any()).Return() + + s.mockStatsReporter.EXPECT(). + ReportFirebaseLatency(gomock.Any(), s.game, gomock.Any()).Return() + + s.mockStatsReporter.EXPECT(). + HandleNotificationSent(s.game, "gcm", "push-game_gcm-single"). + Do(func(game, platform, topic string) { + done <- struct{}{} + }) + + s.mockStatsReporter.EXPECT(). + HandleNotificationSuccess(s.game, "gcm"). + Return() + + go s.handler.HandleResponses() + s.waitGroup.Add(1) + s.handler.HandleMessages(context.Background(), msg) + + timeout := time.NewTimer(5 * time.Second) + select { + case <-done: + case <-timeout.C: + s.Fail("timed out waiting for empty-platform message to be processed") + } + s.waitGroup.Wait() + }) + s.Run("should not lock sendPushConcurrencyControl when sending multiple messages", func() { newMessage := func() kafkaFCMMessage { token := uuid.NewString() diff --git a/interfaces/client.go b/interfaces/client.go index 91b15da7..1ae58006 100644 --- a/interfaces/client.go +++ b/interfaces/client.go @@ -18,6 +18,7 @@ type Message struct { DryRun bool `json:"dry_run,omitempty"` Data Data `json:"data,omitempty"` Notification *Notification `json:"notification,omitempty"` + Platform string `json:"-"` } // Data defines the custom payload of a message. From cbd31eedf194c1ec68a2052f3ffb79b47e893485 Mon Sep 17 00:00:00 2001 From: Lucas Monteiro Date: Tue, 5 May 2026 11:54:30 -0300 Subject: [PATCH 3/4] feat(firebase): build APNS payload for iOS messages toFirebaseMessage branches on Message.Platform: "ios" routes to buildIOSMessage, anything else (including the empty default) keeps the existing Android logic via buildAndroidMessage. buildIOSMessage produces a messaging.APNSConfig with apns-collapse-id, apns-priority and apns-expiration headers (when the corresponding Message fields are set), and an Aps payload covering alert title/body/loc-keys/loc-args, sound, badge (parsed from string to int, silently dropped if not numeric) and content-available for silent pushes. --- extensions/firebase/client/firebase.go | 70 ++++++ .../firebase/client/firebase_apns_test.go | 210 ++++++++++++++++++ 2 files changed, 280 insertions(+) create mode 100644 extensions/firebase/client/firebase_apns_test.go diff --git a/extensions/firebase/client/firebase.go b/extensions/firebase/client/firebase.go index e9d6dc45..b58ae686 100644 --- a/extensions/firebase/client/firebase.go +++ b/extensions/firebase/client/firebase.go @@ -4,6 +4,7 @@ import ( "context" "encoding/json" "errors" + "strconv" "time" firebase "firebase.google.com/go/v4" @@ -94,6 +95,13 @@ func getProjectIDFromJson(jsonStr string) (string, error) { } func toFirebaseMessage(message interfaces.Message) messaging.Message { + if message.Platform == "ios" { + return buildIOSMessage(message) + } + return buildAndroidMessage(message) +} + +func buildAndroidMessage(message interfaces.Message) messaging.Message { firebaseMessage := messaging.Message{ Token: message.To, } @@ -141,6 +149,68 @@ func toFirebaseMessage(message interfaces.Message) messaging.Message { return firebaseMessage } +func buildIOSMessage(message interfaces.Message) messaging.Message { + firebaseMessage := messaging.Message{ + Token: message.To, + } + + if message.Data != nil { + firebaseMessage.Data = toMapString(message.Data) + } + + apns := &messaging.APNSConfig{ + Headers: map[string]string{}, + } + if message.CollapseKey != "" { + apns.Headers["apns-collapse-id"] = message.CollapseKey + } + if message.Priority != "" { + apns.Headers["apns-priority"] = message.Priority + } + if message.TimeToLive != nil { + expiration := time.Now().Add(time.Duration(*message.TimeToLive) * time.Second).Unix() + apns.Headers["apns-expiration"] = strconv.FormatInt(expiration, 10) + } + + aps := &messaging.Aps{ + ContentAvailable: message.ContentAvailable, + } + + if message.Notification != nil { + firebaseMessage.Notification = &messaging.Notification{ + Title: message.Notification.Title, + Body: message.Notification.Body, + ImageURL: message.Notification.ImageUrl, + } + + alert := &messaging.ApsAlert{ + Title: message.Notification.Title, + Body: message.Notification.Body, + LocKey: message.Notification.BodyLocKey, + TitleLocKey: message.Notification.TitleLocKey, + } + if message.Notification.BodyLocArgs != "" { + alert.LocArgs = []string{message.Notification.BodyLocArgs} + } + if message.Notification.TitleLocArgs != "" { + alert.TitleLocArgs = []string{message.Notification.TitleLocArgs} + } + aps.Alert = alert + aps.Sound = message.Notification.Sound + + if message.Notification.Badge != "" { + if badge, err := strconv.Atoi(message.Notification.Badge); err == nil { + aps.Badge = &badge + } + } + } + + apns.Payload = &messaging.APNSPayload{Aps: aps} + firebaseMessage.APNS = apns + + return firebaseMessage +} + func toMapString(data interfaces.Data) map[string]string { result := make(map[string]string) for k, v := range data { diff --git a/extensions/firebase/client/firebase_apns_test.go b/extensions/firebase/client/firebase_apns_test.go new file mode 100644 index 00000000..dcaac9b4 --- /dev/null +++ b/extensions/firebase/client/firebase_apns_test.go @@ -0,0 +1,210 @@ +package client + +import ( + "strconv" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/topfreegames/pusher/interfaces" +) + +func TestToFirebaseMessage_IOSBuildsAPNSOnly(t *testing.T) { + msg := interfaces.Message{ + To: "ios-token", + Platform: "ios", + Notification: &interfaces.Notification{ + Title: "hello", + Body: "world", + Sound: "default", + }, + } + + out := toFirebaseMessage(msg) + + require.NotNil(t, out.APNS) + require.NotNil(t, out.APNS.Payload) + require.NotNil(t, out.APNS.Payload.Aps) + require.NotNil(t, out.APNS.Payload.Aps.Alert) + assert.Equal(t, "hello", out.APNS.Payload.Aps.Alert.Title) + assert.Equal(t, "world", out.APNS.Payload.Aps.Alert.Body) + assert.Equal(t, "default", out.APNS.Payload.Aps.Sound) + assert.Nil(t, out.Android, "iOS message must not populate Android config") +} + +func TestToFirebaseMessage_GCMBuildsAndroidOnly(t *testing.T) { + msg := interfaces.Message{ + To: "android-token", + Platform: "gcm", + Notification: &interfaces.Notification{ + Title: "hello", + Body: "world", + }, + } + + out := toFirebaseMessage(msg) + + require.NotNil(t, out.Android) + require.NotNil(t, out.Android.Notification) + assert.Equal(t, "hello", out.Android.Notification.Title) + assert.Nil(t, out.APNS, "Android message must not populate APNS config") +} + +func TestToFirebaseMessage_EmptyPlatformDefaultsToAndroid(t *testing.T) { + msg := interfaces.Message{ + To: "android-token", + Platform: "", + Notification: &interfaces.Notification{ + Title: "hello", + Body: "world", + }, + } + + out := toFirebaseMessage(msg) + + require.NotNil(t, out.Android) + assert.Nil(t, out.APNS) +} + +func TestBuildIOSMessage_BadgeStringToInt(t *testing.T) { + msg := interfaces.Message{ + To: "ios-token", + Platform: "ios", + Notification: &interfaces.Notification{ + Title: "t", + Body: "b", + Badge: "42", + }, + } + + out := toFirebaseMessage(msg) + + require.NotNil(t, out.APNS.Payload.Aps.Badge) + assert.Equal(t, 42, *out.APNS.Payload.Aps.Badge) +} + +func TestBuildIOSMessage_BadgeInvalidStringIsOmitted(t *testing.T) { + msg := interfaces.Message{ + To: "ios-token", + Platform: "ios", + Notification: &interfaces.Notification{ + Title: "t", + Body: "b", + Badge: "not-a-number", + }, + } + + out := toFirebaseMessage(msg) + + assert.Nil(t, out.APNS.Payload.Aps.Badge) +} + +func TestBuildIOSMessage_BadgeEmptyIsOmitted(t *testing.T) { + msg := interfaces.Message{ + To: "ios-token", + Platform: "ios", + Notification: &interfaces.Notification{ + Title: "t", + Body: "b", + }, + } + + out := toFirebaseMessage(msg) + + assert.Nil(t, out.APNS.Payload.Aps.Badge) +} + +func TestBuildIOSMessage_SilentPush(t *testing.T) { + msg := interfaces.Message{ + To: "ios-token", + Platform: "ios", + ContentAvailable: true, + Data: interfaces.Data{"k": "v"}, + } + + out := toFirebaseMessage(msg) + + require.NotNil(t, out.APNS) + require.NotNil(t, out.APNS.Payload) + require.NotNil(t, out.APNS.Payload.Aps) + assert.True(t, out.APNS.Payload.Aps.ContentAvailable) + assert.Nil(t, out.APNS.Payload.Aps.Alert, "silent push must not include an alert") + assert.Nil(t, out.Notification, "silent push must not include a top-level Notification") + assert.Equal(t, map[string]string{"k": "v"}, out.Data) +} + +func TestBuildIOSMessage_CollapseKeyAndTTLProduceHeaders(t *testing.T) { + ttl := uint(60) + msg := interfaces.Message{ + To: "ios-token", + Platform: "ios", + CollapseKey: "collapse-1", + Priority: "10", + TimeToLive: &ttl, + Notification: &interfaces.Notification{ + Title: "t", + Body: "b", + }, + } + + before := time.Now().Unix() + out := toFirebaseMessage(msg) + after := time.Now().Unix() + + require.NotNil(t, out.APNS) + headers := out.APNS.Headers + assert.Equal(t, "collapse-1", headers["apns-collapse-id"]) + assert.Equal(t, "10", headers["apns-priority"]) + + expStr, ok := headers["apns-expiration"] + require.True(t, ok, "apns-expiration header must be set when TTL provided") + exp, err := strconv.ParseInt(expStr, 10, 64) + require.NoError(t, err) + assert.GreaterOrEqual(t, exp, before+int64(ttl)) + assert.LessOrEqual(t, exp, after+int64(ttl)) +} + +func TestBuildIOSMessage_OmitsHeadersWhenNotProvided(t *testing.T) { + msg := interfaces.Message{ + To: "ios-token", + Platform: "ios", + Notification: &interfaces.Notification{ + Title: "t", + Body: "b", + }, + } + + out := toFirebaseMessage(msg) + + require.NotNil(t, out.APNS) + _, hasCollapse := out.APNS.Headers["apns-collapse-id"] + _, hasPriority := out.APNS.Headers["apns-priority"] + _, hasExpiration := out.APNS.Headers["apns-expiration"] + assert.False(t, hasCollapse) + assert.False(t, hasPriority) + assert.False(t, hasExpiration) +} + +func TestBuildIOSMessage_LocKeysAndArgs(t *testing.T) { + msg := interfaces.Message{ + To: "ios-token", + Platform: "ios", + Notification: &interfaces.Notification{ + BodyLocKey: "body.key", + BodyLocArgs: "arg1", + TitleLocKey: "title.key", + TitleLocArgs: "title-arg", + }, + } + + out := toFirebaseMessage(msg) + + require.NotNil(t, out.APNS.Payload.Aps.Alert) + alert := out.APNS.Payload.Aps.Alert + assert.Equal(t, "body.key", alert.LocKey) + assert.Equal(t, []string{"arg1"}, alert.LocArgs) + assert.Equal(t, "title.key", alert.TitleLocKey) + assert.Equal(t, []string{"title-arg"}, alert.TitleLocArgs) +} + From 9b850a86bc76c4a47420aea5b17c5da2e98a3385 Mon Sep 17 00:00:00 2001 From: Lucas Monteiro Date: Wed, 13 May 2026 12:05:33 -0300 Subject: [PATCH 4/4] fix(firebase): add apns-push-type header to APNS payload Per Apple's APNs spec, every request should declare apns-push-type so APNs can route and validate it. The value must match the payload: "background" for silent (content-available-only) pushes, "alert" otherwise. Background pushes also force apns-priority to 5, since APNs rejects background pushes sent with priority 10. --- extensions/firebase/client/firebase.go | 20 +++++++++++++++++-- .../firebase/client/firebase_apns_test.go | 17 ++++++++++++++++ 2 files changed, 35 insertions(+), 2 deletions(-) diff --git a/extensions/firebase/client/firebase.go b/extensions/firebase/client/firebase.go index b58ae686..aecc43a1 100644 --- a/extensions/firebase/client/firebase.go +++ b/extensions/firebase/client/firebase.go @@ -158,13 +158,19 @@ func buildIOSMessage(message interfaces.Message) messaging.Message { firebaseMessage.Data = toMapString(message.Data) } + pushType := apnsPushType(message) apns := &messaging.APNSConfig{ - Headers: map[string]string{}, + Headers: map[string]string{ + "apns-push-type": pushType, + }, } if message.CollapseKey != "" { apns.Headers["apns-collapse-id"] = message.CollapseKey } - if message.Priority != "" { + if pushType == "background" { + // APNs rejects background pushes with priority 10; default to 5 for silent pushes if not set explicitly + apns.Headers["apns-priority"] = "5" + } else if message.Priority != "" { apns.Headers["apns-priority"] = message.Priority } if message.TimeToLive != nil { @@ -211,6 +217,16 @@ func buildIOSMessage(message interfaces.Message) messaging.Message { return firebaseMessage } +// apnsPushType returns the value for the apns-push-type header. APNs requires +// it to match the payload: "background" for silent pushes (content-available +// only, no alert), "alert" otherwise. +func apnsPushType(message interfaces.Message) string { + if message.Notification == nil && message.ContentAvailable { + return "background" + } + return "alert" +} + func toMapString(data interfaces.Data) map[string]string { result := make(map[string]string) for k, v := range data { diff --git a/extensions/firebase/client/firebase_apns_test.go b/extensions/firebase/client/firebase_apns_test.go index dcaac9b4..a906ab4d 100644 --- a/extensions/firebase/client/firebase_apns_test.go +++ b/extensions/firebase/client/firebase_apns_test.go @@ -30,6 +30,7 @@ func TestToFirebaseMessage_IOSBuildsAPNSOnly(t *testing.T) { assert.Equal(t, "hello", out.APNS.Payload.Aps.Alert.Title) assert.Equal(t, "world", out.APNS.Payload.Aps.Alert.Body) assert.Equal(t, "default", out.APNS.Payload.Aps.Sound) + assert.Equal(t, "alert", out.APNS.Headers["apns-push-type"]) assert.Nil(t, out.Android, "iOS message must not populate Android config") } @@ -132,6 +133,22 @@ func TestBuildIOSMessage_SilentPush(t *testing.T) { assert.Nil(t, out.APNS.Payload.Aps.Alert, "silent push must not include an alert") assert.Nil(t, out.Notification, "silent push must not include a top-level Notification") assert.Equal(t, map[string]string{"k": "v"}, out.Data) + assert.Equal(t, "background", out.APNS.Headers["apns-push-type"]) +} + +func TestBuildIOSMessage_BackgroundForcesPriority5(t *testing.T) { + msg := interfaces.Message{ + To: "ios-token", + Platform: "ios", + ContentAvailable: true, + Priority: "10", + } + + out := toFirebaseMessage(msg) + + assert.Equal(t, "background", out.APNS.Headers["apns-push-type"]) + assert.Equal(t, "5", out.APNS.Headers["apns-priority"], + "background pushes must override priority to 5; APNs rejects priority 10") } func TestBuildIOSMessage_CollapseKeyAndTTLProduceHeaders(t *testing.T) {